ipng/nginx-logtail
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add ASN to logtail, collector, aggregator, frontend and CLI

Browse Source
This commit is contained in:
Pim van Pelt
2026-03-24 02:28:29 +01:00
parent a798bb1d1d
commit 30c8c40157
17 changed files with 566 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
cmd/aggregator/aggregator_test.go
View File

@@ -163,8 +163,8 @@ func TestCacheCoarseRing(t *testing.T) {
func TestCacheQueryTopN(t *testing.T) {
m := NewMerger()
m.Apply(makeSnap("c1", map[string]int64{
st.EncodeTuple(st.Tuple5{Website: "busy.com", Prefix: "1.0.0.0/24", URI: "/", Status: "200"}): 300,
st.EncodeTuple(st.Tuple5{Website: "quiet.com", Prefix: "2.0.0.0/24", URI: "/", Status: "200"}): 50,
st.EncodeTuple(st.Tuple6{Website: "busy.com", Prefix: "1.0.0.0/24", URI: "/", Status: "200"}): 300,
st.EncodeTuple(st.Tuple6{Website: "quiet.com", Prefix: "2.0.0.0/24", URI: "/", Status: "200"}): 50,
}))
cache := NewCache(m, "test")
@@ -181,8 +181,8 @@ func TestCacheQueryTopN(t *testing.T) {
func TestCacheQueryTopNWithFilter(t *testing.T) {
m := NewMerger()
status429 := st.EncodeTuple(st.Tuple5{Website: "example.com", Prefix: "1.0.0.0/24", URI: "/api", Status: "429"})
status200 := st.EncodeTuple(st.Tuple5{Website: "example.com", Prefix: "2.0.0.0/24", URI: "/api", Status: "200"})
status429 := st.EncodeTuple(st.Tuple6{Website: "example.com", Prefix: "1.0.0.0/24", URI: "/api", Status: "429"})
status200 := st.EncodeTuple(st.Tuple6{Website: "example.com", Prefix: "2.0.0.0/24", URI: "/api", Status: "200"})
m.Apply(makeSnap("c1", map[string]int64{status429: 200, status200: 500}))
cache := NewCache(m, "test")
@@ -202,7 +202,7 @@ func TestCacheQueryTrend(t *testing.T) {
for i, count := range []int64{10, 20, 30} {
m.Apply(makeSnap("c1", map[string]int64{
st.EncodeTuple(st.Tuple5{Website: "x.com", Prefix: "1.0.0.0/24", URI: "/", Status: "200"}): count,
st.EncodeTuple(st.Tuple6{Website: "x.com", Prefix: "1.0.0.0/24", URI: "/", Status: "200"}): count,
}))
cache.rotate(now.Add(time.Duration(i) * time.Minute))
}
@@ -270,12 +270,12 @@ func startFakeCollector(t *testing.T, snaps []*pb.Snapshot) string {
func TestGRPCEndToEnd(t *testing.T) {
// Two fake collectors with overlapping labels.
snap1 := makeSnap("col1", map[string]int64{
st.EncodeTuple(st.Tuple5{Website: "busy.com", Prefix: "1.0.0.0/24", URI: "/", Status: "200"}): 500,
st.EncodeTuple(st.Tuple5{Website: "quiet.com", Prefix: "2.0.0.0/24", URI: "/", Status: "429"}): 100,
st.EncodeTuple(st.Tuple6{Website: "busy.com", Prefix: "1.0.0.0/24", URI: "/", Status: "200"}): 500,
st.EncodeTuple(st.Tuple6{Website: "quiet.com", Prefix: "2.0.0.0/24", URI: "/", Status: "429"}): 100,
})
snap2 := makeSnap("col2", map[string]int64{
st.EncodeTuple(st.Tuple5{Website: "busy.com", Prefix: "3.0.0.0/24", URI: "/", Status: "200"}): 300,
st.EncodeTuple(st.Tuple5{Website: "other.com", Prefix: "4.0.0.0/24", URI: "/", Status: "200"}): 50,
st.EncodeTuple(st.Tuple6{Website: "busy.com", Prefix: "3.0.0.0/24", URI: "/", Status: "200"}): 300,
st.EncodeTuple(st.Tuple6{Website: "other.com", Prefix: "4.0.0.0/24", URI: "/", Status: "200"}): 50,
})
addr1 := startFakeCollector(t, []*pb.Snapshot{snap1})
addr2 := startFakeCollector(t, []*pb.Snapshot{snap2})
@@ -388,7 +388,7 @@ func TestGRPCEndToEnd(t *testing.T) {
func TestDegradedCollector(t *testing.T) {
// Start one real and one immediately-gone collector.
snap1 := makeSnap("col1", map[string]int64{
st.EncodeTuple(st.Tuple5{Website: "good.com", Prefix: "1.0.0.0/24", URI: "/", Status: "200"}): 100,
st.EncodeTuple(st.Tuple6{Website: "good.com", Prefix: "1.0.0.0/24", URI: "/", Status: "200"}): 100,
})
addr1 := startFakeCollector(t, []*pb.Snapshot{snap1})
// addr2 points at nothing — connections will fail immediately.

13
cmd/cli/flags.go
View File

@@ -21,6 +21,7 @@ type sharedFlags struct {
websiteRe string // RE2 regex against website
uriRe string // RE2 regex against request URI
isTor string // "", "1" / "!=0" (TOR only), "0" / "!=1" (non-TOR only)
asn string // expression: "12345", "!=65000", ">=1000", etc.
}
// bindShared registers the shared flags on fs and returns a pointer to the
@@ -36,6 +37,7 @@ func bindShared(fs *flag.FlagSet) (*sharedFlags, *string) {
fs.StringVar(&sf.websiteRe, "website-re", "", "filter: RE2 regex against website")
fs.StringVar(&sf.uriRe, "uri-re", "", "filter: RE2 regex against request URI")
fs.StringVar(&sf.isTor, "is-tor", "", "filter: TOR traffic (1 or !=0 = TOR only; 0 or !=1 = non-TOR only)")
fs.StringVar(&sf.asn, "asn", "", "filter: ASN expression (12345, !=65000, >=1000, <64512, …)")
return sf, target
}
@@ -58,7 +60,7 @@ func parseTargets(s string) []string {
}
func buildFilter(sf *sharedFlags) *pb.Filter {
if sf.website == "" && sf.prefix == "" && sf.uri == "" && sf.status == "" && sf.websiteRe == "" && sf.uriRe == "" && sf.isTor == "" {
if sf.website == "" && sf.prefix == "" && sf.uri == "" && sf.status == "" && sf.websiteRe == "" && sf.uriRe == "" && sf.isTor == "" && sf.asn == "" {
return nil
}
f := &pb.Filter{}
@@ -97,6 +99,15 @@ func buildFilter(sf *sharedFlags) *pb.Filter {
fmt.Fprintf(os.Stderr, "--is-tor: invalid value %q; use 1, 0, !=0, or !=1\n", sf.isTor)
os.Exit(1)
}
if sf.asn != "" {
n, op, ok := st.ParseStatusExpr(sf.asn)
if !ok {
fmt.Fprintf(os.Stderr, "--asn: invalid expression %q; use e.g. 12345, !=65000, >=1000, <64512\n", sf.asn)
os.Exit(1)
}
f.AsnNumber = &n
f.AsnOp = op
}
return f
}

23
cmd/collector/parser.go
View File

@@ -3,6 +3,7 @@ package main
import (
"fmt"
"net"
"strconv"
"strings"
)
@@ -13,18 +14,20 @@ type LogRecord struct {
URI string
Status string
IsTor bool
ASN int32
}
// ParseLine parses a tab-separated logtail log line:
//
// $host \t $remote_addr \t $msec \t $request_method \t $request_uri \t $status \t $body_bytes_sent \t $request_time \t $is_tor
// $host \t $remote_addr \t $msec \t $request_method \t $request_uri \t $status \t $body_bytes_sent \t $request_time \t $is_tor \t $asn
//
// The is_tor field (0 or 1) is optional for backward compatibility with
// older log files that omit it; it defaults to false when absent.
// The is_tor (field 9) and asn (field 10) fields are optional for backward
// compatibility with older log files that omit them; they default to false/0
// when absent.
// Returns false for lines with fewer than 8 fields.
func ParseLine(line string, v4bits, v6bits int) (LogRecord, bool) {
// SplitN caps allocations; we need up to 9 fields.
fields := strings.SplitN(line, "\t", 9)
// SplitN caps allocations; we need up to 10 fields.
fields := strings.SplitN(line, "\t", 10)
if len(fields) < 8 {
return LogRecord{}, false
}
@@ -39,7 +42,14 @@ func ParseLine(line string, v4bits, v6bits int) (LogRecord, bool) {
return LogRecord{}, false
}
isTor := len(fields) == 9 && fields[8] == "1"
isTor := len(fields) >= 9 && fields[8] == "1"
var asn int32
if len(fields) == 10 {
if n, err := strconv.ParseInt(fields[9], 10, 32); err == nil {
asn = int32(n)
}
}
return LogRecord{
Website: fields[0],
@@ -47,6 +57,7 @@ func ParseLine(line string, v4bits, v6bits int) (LogRecord, bool) {
URI: uri,
Status: fields[5],
IsTor: isTor,
ASN: asn,
}, true
}

52
cmd/collector/parser_test.go
View File

@@ -108,6 +108,58 @@ func TestParseLine(t *testing.T) {
IsTor: false,
},
},
{
name: "asn field parsed",
line: "asn.example.com\t1.2.3.4\t0\tGET\t/\t200\t0\t0.001\t0\t12345",
wantOK: true,
want: LogRecord{
Website: "asn.example.com",
ClientPrefix: "1.2.3.0/24",
URI: "/",
Status: "200",
IsTor: false,
ASN: 12345,
},
},
{
name: "asn field with is_tor=1",
line: "both.example.com\t1.2.3.4\t0\tGET\t/\t200\t0\t0.001\t1\t65535",
wantOK: true,
want: LogRecord{
Website: "both.example.com",
ClientPrefix: "1.2.3.0/24",
URI: "/",
Status: "200",
IsTor: true,
ASN: 65535,
},
},
{
name: "missing asn field defaults to 0 (backward compat)",
line: "noasn.example.com\t1.2.3.4\t0\tGET\t/\t200\t0\t0.001\t1",
wantOK: true,
want: LogRecord{
Website: "noasn.example.com",
ClientPrefix: "1.2.3.0/24",
URI: "/",
Status: "200",
IsTor: true,
ASN: 0,
},
},
{
name: "invalid asn field defaults to 0",
line: "badann.example.com\t1.2.3.4\t0\tGET\t/\t200\t0\t0.001\t0\tnot-a-number",
wantOK: true,
want: LogRecord{
Website: "badann.example.com",
ClientPrefix: "1.2.3.0/24",
URI: "/",
Status: "200",
IsTor: false,
ASN: 0,
},
},
}
for _, tc := range tests {

8
cmd/collector/store.go
View File

@@ -15,7 +15,7 @@ type Store struct {
source string
// live map — written only by the Run goroutine; no locking needed on writes
live map[st.Tuple5]int64
live map[st.Tuple6]int64
liveLen int
// ring buffers — protected by mu for reads
@@ -36,7 +36,7 @@ type Store struct {
func NewStore(source string) *Store {
return &Store{
source: source,
live: make(map[st.Tuple5]int64, liveMapCap),
live: make(map[st.Tuple6]int64, liveMapCap),
subs: make(map[chan st.Snapshot]struct{}),
}
}
@@ -44,7 +44,7 @@ func NewStore(source string) *Store {
// ingest records one log record into the live map.
// Must only be called from the Run goroutine.
func (s *Store) ingest(r LogRecord) {
key := st.Tuple5{Website: r.Website, Prefix: r.ClientPrefix, URI: r.URI, Status: r.Status, IsTor: r.IsTor}
key := st.Tuple6{Website: r.Website, Prefix: r.ClientPrefix, URI: r.URI, Status: r.Status, IsTor: r.IsTor, ASN: r.ASN}
if _, exists := s.live[key]; !exists {
if s.liveLen >= liveMapCap {
return
@@ -77,7 +77,7 @@ func (s *Store) rotate(now time.Time) {
}
s.mu.Unlock()
s.live = make(map[st.Tuple5]int64, liveMapCap)
s.live = make(map[st.Tuple6]int64, liveMapCap)
s.liveLen = 0
s.broadcast(fine)

29
cmd/frontend/filter.go
View File

@@ -126,8 +126,20 @@ func applyTerm(term string, fs *filterState) error {
} else {
fs.IsTor = "0"
}
case "asn":
if op == "~=" {
return fmt.Errorf("asn does not support ~=; use =, !=, >=, >, <=, <")
}
expr := op + value
if op == "=" {
expr = value
}
if _, _, ok := st.ParseStatusExpr(expr); !ok {
return fmt.Errorf("invalid asn expression %q", expr)
}
fs.ASN = expr
default:
return fmt.Errorf("unknown field %q; valid: status, website, uri, prefix, is_tor", field)
return fmt.Errorf("unknown field %q; valid: status, website, uri, prefix, is_tor, asn", field)
}
return nil
}
@@ -167,9 +179,24 @@ func FilterExprString(f filterState) string {
if f.IsTor != "" {
parts = append(parts, "is_tor="+f.IsTor)
}
if f.ASN != "" {
parts = append(parts, asnTermStr(f.ASN))
}
return strings.Join(parts, " AND ")
}
// asnTermStr converts a stored ASN expression (">=1000", "12345") to a
// full filter term ("asn>=1000", "asn=12345").
func asnTermStr(expr string) string {
if expr == "" {
return ""
}
if len(expr) > 0 && (expr[0] == '!' || expr[0] == '>' || expr[0] == '<') {
return "asn" + expr
}
return "asn=" + expr
}
// statusTermStr converts a stored status expression (">=400", "200") to a
// full filter term ("status>=400", "status=200").
func statusTermStr(expr string) string {

74
cmd/frontend/filter_test.go
View File

@@ -258,3 +258,77 @@ func TestFilterExprRoundTrip(t *testing.T) {
}
}
}
func TestParseAsnEQ(t *testing.T) {
fs, err := ParseFilterExpr("asn=12345")
if err != nil || fs.ASN != "12345" {
t.Fatalf("got err=%v fs=%+v", err, fs)
}
}
func TestParseAsnNE(t *testing.T) {
fs, err := ParseFilterExpr("asn!=65000")
if err != nil || fs.ASN != "!=65000" {
t.Fatalf("got err=%v fs=%+v", err, fs)
}
}
func TestParseAsnGE(t *testing.T) {
fs, err := ParseFilterExpr("asn>=1000")
if err != nil || fs.ASN != ">=1000" {
t.Fatalf("got err=%v fs=%+v", err, fs)
}
}
func TestParseAsnLT(t *testing.T) {
fs, err := ParseFilterExpr("asn<64512")
if err != nil || fs.ASN != "<64512" {
t.Fatalf("got err=%v fs=%+v", err, fs)
}
}
func TestParseAsnRegexRejected(t *testing.T) {
_, err := ParseFilterExpr("asn~=123")
if err == nil {
t.Fatal("expected error for asn~=")
}
}
func TestParseAsnInvalidExpr(t *testing.T) {
_, err := ParseFilterExpr("asn=notanumber")
if err == nil {
t.Fatal("expected error for non-numeric ASN")
}
}
func TestFilterExprStringASN(t *testing.T) {
s := FilterExprString(filterState{ASN: "12345"})
if s != "asn=12345" {
t.Fatalf("got %q", s)
}
s = FilterExprString(filterState{ASN: ">=1000"})
if s != "asn>=1000" {
t.Fatalf("got %q", s)
}
}
func TestFilterExprRoundTripASN(t *testing.T) {
cases := []filterState{
{ASN: "12345"},
{ASN: "!=65000"},
{ASN: ">=1000"},
{ASN: "<64512"},
{Status: ">=400", ASN: "12345"},
}
for _, fs := range cases {
expr := FilterExprString(fs)
fs2, err := ParseFilterExpr(expr)
if err != nil {
t.Errorf("round-trip parse error for %+v → %q: %v", fs, expr, err)
continue
}
if fs2 != fs {
t.Errorf("round-trip mismatch: %+v → %q → %+v", fs, expr, fs2)
}
}
}

11
cmd/frontend/frontend_test.go
View File

@@ -220,8 +220,17 @@ func TestDrillURL(t *testing.T) {
if !strings.Contains(u, "f_status=429") {
t.Errorf("drill from status: missing f_status in %q", u)
}
if !strings.Contains(u, "by=asn") {
t.Errorf("drill from status: expected next by=asn in %q", u)
}
p.GroupByS = "asn"
u = p.drillURL("12345")
if !strings.Contains(u, "f_asn=12345") {
t.Errorf("drill from asn: missing f_asn in %q", u)
}
if !strings.Contains(u, "by=website") {
t.Errorf("drill from status: expected cycle back to by=website in %q", u)
t.Errorf("drill from asn: expected cycle back to by=website in %q", u)
}
}

31
cmd/frontend/handler.go
View File

@@ -54,6 +54,7 @@ type filterState struct {
WebsiteRe string // RE2 regex against website
URIRe string // RE2 regex against request URI
IsTor string // "", "1" (TOR only), "0" (non-TOR only)
ASN string // expression: "12345", "!=65000", ">=1000", etc.
}
// QueryParams holds all parsed URL parameters for one page request.
@@ -91,7 +92,7 @@ var windowSpecs = []struct{ s, label string }{
}
var groupBySpecs = []struct{ s, label string }{
{"website", "website"}, {"prefix", "prefix"}, {"uri", "uri"}, {"status", "status"},
{"website", "website"}, {"asn", "asn"}, {"prefix", "prefix"}, {"status", "status"}, {"uri", "uri"},
}
func parseWindowString(s string) (pb.Window, string) {
@@ -121,6 +122,8 @@ func parseGroupByString(s string) (pb.GroupBy, string) {
return pb.GroupBy_REQUEST_URI, "uri"
case "status":
return pb.GroupBy_HTTP_RESPONSE, "status"
case "asn":
return pb.GroupBy_ASN_NUMBER, "asn"
default:
return pb.GroupBy_WEBSITE, "website"
}
@@ -159,12 +162,13 @@ func (h *Handler) parseParams(r *http.Request) QueryParams {
WebsiteRe: q.Get("f_website_re"),
URIRe: q.Get("f_uri_re"),
IsTor: q.Get("f_is_tor"),
ASN: q.Get("f_asn"),
},
}
}
func buildFilter(f filterState) *pb.Filter {
if f.Website == "" && f.Prefix == "" && f.URI == "" && f.Status == "" && f.WebsiteRe == "" && f.URIRe == "" && f.IsTor == "" {
if f.Website == "" && f.Prefix == "" && f.URI == "" && f.Status == "" && f.WebsiteRe == "" && f.URIRe == "" && f.IsTor == "" && f.ASN == "" {
return nil
}
out := &pb.Filter{}
@@ -195,6 +199,12 @@ func buildFilter(f filterState) *pb.Filter {
case "0":
out.Tor = pb.TorFilter_TOR_NO
}
if f.ASN != "" {
if n, op, ok := st.ParseStatusExpr(f.ASN); ok {
out.AsnNumber = &n
out.AsnOp = op
}
}
return out
}
@@ -226,6 +236,9 @@ func (p QueryParams) toValues() url.Values {
if p.Filter.IsTor != "" {
v.Set("f_is_tor", p.Filter.IsTor)
}
if p.Filter.ASN != "" {
v.Set("f_asn", p.Filter.ASN)
}
return v
}
@@ -247,7 +260,7 @@ func (p QueryParams) buildURL(overrides map[string]string) string {
func (p QueryParams) clearFilterURL() string {
return p.buildURL(map[string]string{
"f_website": "", "f_prefix": "", "f_uri": "", "f_status": "",
"f_website_re": "", "f_uri_re": "",
"f_website_re": "", "f_uri_re": "", "f_is_tor": "", "f_asn": "",
})
}
@@ -260,7 +273,9 @@ func nextGroupBy(s string) string {
return "uri"
case "uri":
return "status"
default: // status → back to website
case "status":
return "asn"
default: // asn → back to website
return "website"
}
}
@@ -276,6 +291,8 @@ func groupByFilterKey(s string) string {
return "f_uri"
case "status":
return "f_status"
case "asn":
return "f_asn"
default:
return "f_website"
}
@@ -338,6 +355,12 @@ func buildCrumbs(p QueryParams) []Crumb {
RemoveURL: p.buildURL(map[string]string{"f_is_tor": ""}),
})
}
if p.Filter.ASN != "" {
crumbs = append(crumbs, Crumb{
Text: asnTermStr(p.Filter.ASN),
RemoveURL: p.buildURL(map[string]string{"f_asn": ""}),
})
}
return crumbs
}

2
cmd/frontend/templates/index.html
View File

@@ -32,7 +32,7 @@
<input type="hidden" name="w" value="{{.Params.WindowS}}">
<input type="hidden" name="by" value="{{.Params.GroupByS}}">
<input type="hidden" name="n" value="{{.Params.N}}">
<input class="filter-input" type="text" name="q" value="{{.FilterExpr}}" placeholder="status>=400 AND website~=gouda.* AND uri~=^/api/ AND is_tor=1">
<input class="filter-input" type="text" name="q" value="{{.FilterExpr}}" placeholder="status>=400 AND website~=gouda.* AND uri~=^/ct/v1/ AND is_tor=0 AND asn=8298">
<button type="submit">filter</button>
{{- if .FilterExpr}} <a class="clear" href="{{.ClearFilterURL}}">× clear</a>{{end}}
</form>