Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
8.0 kB · 269 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270package bunny
import ( "bufio" "compress/gzip" "context" "encoding/json" "fmt" "io" "net/http" "net/url" "path" "strconv" "strings" "time"
"stream.place/streamplace/pkg/aqhttp" "stream.place/streamplace/pkg/cdn" "stream.place/streamplace/pkg/log")
// Request is the provider-neutral record type this package emits.// Aliased so the parser tests don't have to import pkg/cdn.type Request = cdn.Request
// ParseLine decodes one line of bunny's pipe-delimited access-log// format://// cache-status|status|timestamp-ms|bytes-sent|pull-zone-id|remote-ip|// referer|url|edge-location|user-agent|request-id|country-code//// Bunny strips '|' from the user-controllable fields, so a naive split// is safe. Lines with fewer than 12 fields, or an unparseable status,// timestamp or byte count, are rejected. Extra trailing fields (a// future format extension) are ignored.func ParseLine(line string) (Request, error) { line = strings.TrimRight(line, "\r\n") f := strings.Split(line, "|") if len(f) < 12 { return Request{}, fmt.Errorf("bunny log: %d fields, want 12", len(f)) } status, err := strconv.Atoi(f[1]) if err != nil { return Request{}, fmt.Errorf("bunny log: status %q: %w", f[1], err) } ms, err := strconv.ParseInt(f[2], 10, 64) if err != nil { return Request{}, fmt.Errorf("bunny log: timestamp %q: %w", f[2], err) } bytes, err := strconv.ParseInt(f[3], 10, 64) if err != nil { return Request{}, fmt.Errorf("bunny log: bytes %q: %w", f[3], err) } return Request{ Time: time.UnixMilli(ms).UTC(), Status: status, BytesSent: bytes, RemoteIP: f[5], URL: f[7], }, nil}
// LogStorage implements cdn.LogSource over a bunny Edge Storage zone// that a pull zone's Permanent Log Storage writes into. Bunny files// parts as//// pullzone-logs/<pull-zone>/<YYYY>/<MM>/<dd>_<worker>-<part>-<rand>.gzip//// Each part is immutable once it lands (parts close on size, line// count, age or the UTC midnight boundary and are never rewritten),// which is what lets the ingester treat "seen this path" as "done".type LogStorage struct { // Endpoint is the storage region's API base, e.g. // https://storage.bunnycdn.com or https://ny.storage.bunnycdn.com. Endpoint string // Zone is the storage zone name. Zone string // AccessKey is the storage zone's password (read-only is enough). AccessKey string // PullZone is the pull zone name as it appears in the log path. PullZone string // HTTPClient overrides the outbound client (tests). Defaults to // aqhttp.TrustedClient: the endpoint is operator-configured. HTTPClient *http.Client // Now overrides the clock (tests). Now func() time.Time}
// storageObject is the subset of bunny's List Files response we read.type storageObject struct { ObjectName string `json:"ObjectName"` IsDirectory bool `json:"IsDirectory"` Length int64 `json:"Length"`}
func (s *LogStorage) client() *http.Client { if s.HTTPClient != nil { return s.HTTPClient } return &aqhttp.TrustedClient}
func (s *LogStorage) now() time.Time { if s.Now != nil { return s.Now() } return time.Now()}
// logPrefix is the directory all of this pull zone's parts live under.func (s *LogStorage) logPrefix() string { return "pullzone-logs/" + s.PullZone}
func (s *LogStorage) objectURL(p string) (string, error) { base, err := url.Parse(strings.TrimRight(s.Endpoint, "/")) if err != nil || base.Host == "" { return "", fmt.Errorf("bunny storage: bad endpoint %q", s.Endpoint) } base.Path = path.Join(base.Path, s.Zone, p) // Directory listings need the trailing slash; path.Join drops it. if strings.HasSuffix(p, "/") { base.Path += "/" } return base.String(), nil}
func (s *LogStorage) get(ctx context.Context, p string) (*http.Response, error) { u, err := s.objectURL(p) if err != nil { return nil, err } req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil) if err != nil { return nil, err } req.Header.Set("AccessKey", s.AccessKey) req.Header.Set("Accept", "*/*") resp, err := s.client().Do(req) if err != nil { return nil, fmt.Errorf("bunny storage: GET %s: %w", p, err) } return resp, nil}
// ListParts walks the month directories from since's UTC month through// the current month and returns every part whose day-of-month prefix// falls on or after since's UTC date.func (s *LogStorage) ListParts(ctx context.Context, since time.Time) ([]cdn.Part, error) { since = since.UTC() sinceDay := time.Date(since.Year(), since.Month(), since.Day(), 0, 0, 0, 0, time.UTC) now := s.now().UTC() var parts []cdn.Part for m := time.Date(since.Year(), since.Month(), 1, 0, 0, 0, 0, time.UTC); !m.After(now); m = m.AddDate(0, 1, 0) { dir := fmt.Sprintf("%s/%04d/%02d/", s.logPrefix(), m.Year(), int(m.Month())) objs, err := s.list(ctx, dir) if err != nil { return nil, err } for _, o := range objs { if o.IsDirectory { continue } day, ok := partDay(m, o.ObjectName) if !ok { log.Debug(ctx, "bunny storage: skipping unrecognized object", "dir", dir, "name", o.ObjectName) continue } if day.Before(sinceDay) { continue } parts = append(parts, cdn.Part{ ID: dir + o.ObjectName, Day: day, Size: o.Length, }) } } return parts, nil}
// partDay extracts the UTC date from a part filename like// `06_abc123-0-1a2b3c4d.gzip` inside the given month directory.func partDay(month time.Time, name string) (time.Time, bool) { if !strings.HasSuffix(name, ".gzip") && !strings.HasSuffix(name, ".gz") { return time.Time{}, false } i := strings.IndexByte(name, '_') if i <= 0 { return time.Time{}, false } d, err := strconv.Atoi(name[:i]) if err != nil || d < 1 || d > 31 { return time.Time{}, false } return time.Date(month.Year(), month.Month(), d, 0, 0, 0, 0, time.UTC), true}
func (s *LogStorage) list(ctx context.Context, dir string) ([]storageObject, error) { resp, err := s.get(ctx, dir) if err != nil { return nil, err } defer resp.Body.Close() if resp.StatusCode == http.StatusNotFound { // A month with no logs yet (or a pull zone that hasn't // written anything) is not an error. return nil, nil } if resp.StatusCode != http.StatusOK { return nil, fmt.Errorf("bunny storage: list %s: HTTP %d", dir, resp.StatusCode) } var objs []storageObject if err := json.NewDecoder(resp.Body).Decode(&objs); err != nil { return nil, fmt.Errorf("bunny storage: decode listing %s: %w", dir, err) } return objs, nil}
// maxLineBytes bounds a single log line. Bunny lines are a few hundred// bytes; user-agent + referer are the only unbounded fields.const maxLineBytes = 1 << 20
// ReadPart downloads the gzip part and streams parsed records. Lines// that fail to parse are skipped and counted; the count is logged at// the end so a format change is visible without failing the whole// part.func (s *LogStorage) ReadPart(ctx context.Context, part cdn.Part, emit func(Request) error) error { resp, err := s.get(ctx, part.ID) if err != nil { return err } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { return fmt.Errorf("bunny storage: read %s: HTTP %d", part.ID, resp.StatusCode) } gz, err := gzip.NewReader(resp.Body) if err != nil { return fmt.Errorf("bunny storage: gunzip %s: %w", part.ID, err) } defer gz.Close() return readLines(ctx, gz, part.ID, emit)}
func readLines(ctx context.Context, r io.Reader, name string, emit func(Request) error) error { sc := bufio.NewScanner(r) sc.Buffer(make([]byte, 64*1024), maxLineBytes) var bad int for sc.Scan() { if len(sc.Bytes()) == 0 { continue } req, err := ParseLine(sc.Text()) if err != nil { bad++ continue } if err := emit(req); err != nil { return err } } if bad > 0 { log.Warn(ctx, "bunny storage: skipped unparseable log lines", "part", name, "lines", bad) } if err := sc.Err(); err != nil { return fmt.Errorf("bunny storage: read %s: %w", name, err) } return nil}