Enable lame-duck on exit, so systemctl can 'restart' collectors

Add is_tor plumbing from collector->aggregator->frontend/cli
Auto-rediscover new glob patterns
2026-03-23 22:18:05 +01:00 · 2026-03-23 22:17:39 +01:00 · 2026-03-23 20:39:12 +01:00
23 changed files with 1920 additions and 241 deletions
--- a/README.md
+++ b/README.md
@@ -38,7 +38,7 @@ nginx-logtail/
 │       └── logtail_grpc.pb.go         # generated: service stubs
 ├── internal/
 │   └── store/
-│       └── store.go                   # shared types: Tuple4, Entry, Snapshot, ring helpers
+│       └── store.go                   # shared types: Tuple5, Entry, Snapshot, ring helpers
 └── cmd/
    ├── collector/
    │   ├── main.go
@@ -76,7 +76,7 @@ nginx-logtail/

 ## Data Model

-The core unit is a **count keyed by four dimensions**:
+The core unit is a **count keyed by five dimensions**:

 | Field             | Description                                          | Example           |
 |-------------------|------------------------------------------------------|-------------------|
@@ -84,6 +84,7 @@ The core unit is a **count keyed by four dimensions**:
 | `client_prefix`   | client IP truncated to /24 IPv4 or /48 IPv6          | `1.2.3.0/24`      |
 | `http_request_uri`| `$request_uri` path only — query string stripped     | `/api/v1/search`  |
 | `http_response`   | HTTP status code                                     | `429`             |
+| `is_tor`          | whether the client IP is a TOR exit node             | `1`               |

 ## Time Windows & Tiered Ring Buffers

@@ -110,8 +111,8 @@ Every 5 minutes: merge last 5 fine snapshots → top-5K → append to coarse rin

 ## Memory Budget (Collector, target ≤ 1 GB)

-Entry size: ~30 B website + ~15 B prefix + ~50 B URI + 3 B status + 8 B count + ~80 B Go map
-overhead ≈ **~186 bytes per entry**.
+Entry size: ~30 B website + ~15 B prefix + ~50 B URI + 3 B status + 1 B is_tor + 8 B count + ~80 B Go map
+overhead ≈ **~187 bytes per entry**.

 | Structure               | Entries     | Size        |
 |-------------------------|-------------|-------------|
@@ -151,7 +152,8 @@ and does not change any existing interface.
 ## Protobuf API (`proto/logtail.proto`)

 ```protobuf
-enum StatusOp { EQ = 0; NE = 1; GT = 2; GE = 3; LT = 4; LE = 5; }
+enum TorFilter { TOR_ANY = 0; TOR_YES = 1; TOR_NO = 2; }
+enum StatusOp  { EQ = 0; NE = 1; GT = 2; GE = 3; LT = 4; LE = 5; }

 message Filter {
  optional string website          = 1;
@@ -161,6 +163,7 @@ message Filter {
  StatusOp        status_op        = 5;  // comparison operator for http_response
  optional string website_regex    = 6;  // RE2 regex against website
  optional string uri_regex        = 7;  // RE2 regex against http_request_uri
+  TorFilter       tor              = 8;  // TOR_ANY (default) / TOR_YES / TOR_NO
 }

 enum GroupBy { WEBSITE = 0; CLIENT_PREFIX = 1; REQUEST_URI = 2; HTTP_RESPONSE = 3; }
@@ -217,7 +220,7 @@ service LogtailService {
 - Parses the fixed **logtail** nginx log format — tab-separated, fixed field order, no quoting:

  ```nginx
-  log_format logtail '$host\t$remote_addr\t$msec\t$request_method\t$request_uri\t$status\t$body_bytes_sent\t$request_time';
+  log_format logtail '$host\t$remote_addr\t$msec\t$request_method\t$request_uri\t$status\t$body_bytes_sent\t$request_time\t$is_tor';
  ```

  | # | Field             | Used for         |
@@ -230,16 +233,19 @@ service LogtailService {
  | 5 | `$status`         | http_response    |
  | 6 | `$body_bytes_sent`| (discarded)      |
  | 7 | `$request_time`   | (discarded)      |
+  | 8 | `$is_tor`         | is_tor           |

- `strings.SplitN(line, "\t", 8)` — ~50 ns/line. No regex.
+- `strings.SplitN(line, "\t", 9)` — ~50 ns/line. No regex.
 - `$request_uri`: query string discarded at first `?`.
 - `$remote_addr`: truncated to /24 (IPv4) or /48 (IPv6); prefix lengths configurable via flags.
+- `$is_tor`: `1` if the client IP is a TOR exit node, `0` otherwise. Field is optional — lines
+  with exactly 8 fields (old format) are accepted and default to `is_tor=false`.
 - Lines with fewer than 8 fields are silently skipped.

 ### store.go
 - **Single aggregator goroutine** reads from the channel and updates the live map — no locking on
  the hot path. At 10 K lines/s the goroutine uses <1% CPU.
- Live map: `map[Tuple4]int64`, hard-capped at 100 K entries (new keys dropped when full).
+- Live map: `map[Tuple5]int64`, hard-capped at 100 K entries (new keys dropped when full).
 - **Minute ticker**: heap-selects top-50K entries, writes snapshot to fine ring, resets live map.
 - Every 5 fine ticks: merge last 5 fine snapshots → top-5K → write to coarse ring.
 - **TopN query**: RLock ring, sum bucket range, apply filter, group by dimension, heap-select top N.
@@ -291,7 +297,7 @@ service LogtailService {

 ### handler.go
 - All filter state in the **URL query string**: `w` (window), `by` (group_by), `f_website`,
-  `f_prefix`, `f_uri`, `f_status`, `f_website_re`, `f_uri_re`, `n`, `target`. No server-side
+  `f_prefix`, `f_uri`, `f_status`, `f_website_re`, `f_uri_re`, `f_is_tor`, `n`, `target`. No server-side
  session — URLs are shareable and bookmarkable; multiple operators see independent views.
 - **Filter expression box**: a `q=` parameter carries a mini filter language
  (`status>=400 AND website~=gouda.* AND uri~=^/api/`). On submission the handler parses it
@@ -340,16 +346,17 @@ logtail-cli targets [flags]   list targets known to the queried endpoint

 **Shared** (all subcommands):

-| Flag         | Default          | Description                                              |
-|--------------|------------------|----------------------------------------------------------|
-| `--target`   | `localhost:9090` | Comma-separated `host:port` list; fan-out to all         |
-| `--json`     | false            | Emit newline-delimited JSON instead of a table           |
-| `--website`  | —                | Filter: website                                          |
-| `--prefix`   | —                | Filter: client prefix                                    |
-| `--uri`      | —                | Filter: request URI                                      |
-| `--status`   | —                | Filter: HTTP status expression (`200`, `!=200`, `>=400`, `<500`, …) |
-| `--website-re`| —               | Filter: RE2 regex against website                        |
-| `--uri-re`   | —                | Filter: RE2 regex against request URI                    |
+| Flag          | Default          | Description                                              |
+|---------------|------------------|----------------------------------------------------------|
+| `--target`    | `localhost:9090` | Comma-separated `host:port` list; fan-out to all         |
+| `--json`      | false            | Emit newline-delimited JSON instead of a table           |
+| `--website`   | —                | Filter: website                                          |
+| `--prefix`    | —                | Filter: client prefix                                    |
+| `--uri`       | —                | Filter: request URI                                      |
+| `--status`    | —                | Filter: HTTP status expression (`200`, `!=200`, `>=400`, `<500`, …) |
+| `--website-re`| —                | Filter: RE2 regex against website                        |
+| `--uri-re`    | —                | Filter: RE2 regex against request URI                    |
+| `--is-tor`    | —                | Filter: TOR traffic (`1` or `!=0` = TOR only; `0` or `!=1` = non-TOR only) |

 **`topn` only**: `--n 10`, `--window 5m`, `--group-by website`

@@ -381,7 +388,7 @@ with a non-zero code on gRPC error.
 | Tick-based cache rotation in aggregator | Ring stays on the same 1-min cadence regardless of collector count |
 | Degraded collector zeroing | Stale counts from failed collectors don't accumulate in the merged view |
 | Same `LogtailService` for collector and aggregator | CLI and frontend work with either; no special-casing |
-| `internal/store` shared package | ~200 lines of ring-buffer logic shared between collector and aggregator |
+| `internal/store` shared package | ring-buffer, `Tuple5` encoding, and filter logic shared between collector and aggregator |
 | Filter state in URL, not session cookie | Multiple concurrent operators; shareable/bookmarkable URLs |
 | Query strings stripped at ingest | Major cardinality reduction; prevents URI explosion under attack |
 | No persistent storage | Simplicity; acceptable for ops dashboards (restart = lose history) |
@@ -393,4 +400,4 @@ with a non-zero code on gRPC error.
 | Status filter as expression string (`!=200`, `>=400`) | Operator-friendly; parsed once at query boundary, encoded as `(int32, StatusOp)` in proto |
 | Regex filters compiled once per query (`CompiledFilter`) | Up to 288 × 5 000 per-entry calls — compiling per-entry would dominate query latency |
 | Filter expression box (`q=`) redirects to canonical URL | Filter state stays in individual `f_*` params; URLs remain shareable and bookmarkable |
-| `ListTargets` + frontend source picker (no Tuple5) | "Which nginx is busiest?" answered by switching `target=` to a collector; no data model changes, no extra memory |
+| `ListTargets` + frontend source picker | "Which nginx is busiest?" answered by switching `target=` to a collector; no data model changes, no extra memory |
--- a/cmd/aggregator/aggregator_test.go
+++ b/cmd/aggregator/aggregator_test.go
@@ -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.Tuple4{"busy.com", "1.0.0.0/24", "/", "200"}): 300,
-		st.EncodeTuple(st.Tuple4{"quiet.com", "2.0.0.0/24", "/", "200"}): 50,
+		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,
 	}))

 	cache := NewCache(m, "test")
@@ -181,8 +181,8 @@ func TestCacheQueryTopN(t *testing.T) {

 func TestCacheQueryTopNWithFilter(t *testing.T) {
 	m := NewMerger()
-	status429 := st.EncodeTuple(st.Tuple4{"example.com", "1.0.0.0/24", "/api", "429"})
-	status200 := st.EncodeTuple(st.Tuple4{"example.com", "2.0.0.0/24", "/api", "200"})
+	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"})
 	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.Tuple4{"x.com", "1.0.0.0/24", "/", "200"}): count,
+			st.EncodeTuple(st.Tuple5{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.Tuple4{"busy.com", "1.0.0.0/24", "/", "200"}):  500,
-		st.EncodeTuple(st.Tuple4{"quiet.com", "2.0.0.0/24", "/", "429"}): 100,
+		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,
 	})
 	snap2 := makeSnap("col2", map[string]int64{
-		st.EncodeTuple(st.Tuple4{"busy.com", "3.0.0.0/24", "/", "200"}):  300,
-		st.EncodeTuple(st.Tuple4{"other.com", "4.0.0.0/24", "/", "200"}): 50,
+		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,
 	})
 	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.Tuple4{"good.com", "1.0.0.0/24", "/", "200"}): 100,
+		st.EncodeTuple(st.Tuple5{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.
--- a/cmd/cli/flags.go
+++ b/cmd/cli/flags.go
@@ -20,6 +20,7 @@ type sharedFlags struct {
 	status    string // expression: "200", "!=200", ">=400", etc.
 	websiteRe string // RE2 regex against website
 	uriRe     string // RE2 regex against request URI
+	isTor     string // "", "1" / "!=0" (TOR only), "0" / "!=1" (non-TOR only)
 }

 // bindShared registers the shared flags on fs and returns a pointer to the
@@ -34,6 +35,7 @@ func bindShared(fs *flag.FlagSet) (*sharedFlags, *string) {
 	fs.StringVar(&sf.status, "status", "", "filter: HTTP status expression (200, !=200, >=400, <500, …)")
 	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)")
 	return sf, target
 }

@@ -56,7 +58,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 == "" {
+	if sf.website == "" && sf.prefix == "" && sf.uri == "" && sf.status == "" && sf.websiteRe == "" && sf.uriRe == "" && sf.isTor == "" {
 		return nil
 	}
 	f := &pb.Filter{}
@@ -84,6 +86,17 @@ func buildFilter(sf *sharedFlags) *pb.Filter {
 	if sf.uriRe != "" {
 		f.UriRegex = &sf.uriRe
 	}
+	switch sf.isTor {
+	case "1", "!=0":
+		f.Tor = pb.TorFilter_TOR_YES
+	case "0", "!=1":
+		f.Tor = pb.TorFilter_TOR_NO
+	case "":
+		// no filter
+	default:
+		fmt.Fprintf(os.Stderr, "--is-tor: invalid value %q; use 1, 0, !=0, or !=1\n", sf.isTor)
+		os.Exit(1)
+	}
 	return f
 }

--- a/cmd/collector/main.go
+++ b/cmd/collector/main.go
@@ -11,30 +11,27 @@ import (
 	"path/filepath"
 	"strings"
 	"syscall"
+	"time"

 	pb "git.ipng.ch/ipng/nginx-logtail/proto/logtailpb"
 	"google.golang.org/grpc"
 )

 func main() {
-	listen   := flag.String("listen", ":9090", "gRPC listen address")
-	logPaths := flag.String("logs", "", "comma-separated log file paths/globs to tail")
-	logsFile := flag.String("logs-file", "", "file containing one log path/glob per line")
-	source   := flag.String("source", hostname(), "name for this collector (default: hostname)")
-	v4prefix := flag.Int("v4prefix", 24, "IPv4 prefix length for client bucketing")
-	v6prefix := flag.Int("v6prefix", 48, "IPv6 prefix length for client bucketing")
+	listen       := flag.String("listen", ":9090", "gRPC listen address")
+	logPaths     := flag.String("logs", "", "comma-separated log file paths/globs to tail")
+	logsFile     := flag.String("logs-file", "", "file containing one log path/glob per line")
+	source       := flag.String("source", hostname(), "name for this collector (default: hostname)")
+	v4prefix     := flag.Int("v4prefix", 24, "IPv4 prefix length for client bucketing")
+	v6prefix     := flag.Int("v6prefix", 48, "IPv6 prefix length for client bucketing")
+	scanInterval := flag.Duration("scan-interval", 10*time.Second, "how often to rescan glob patterns for new/removed files")
 	flag.Parse()

 	patterns := collectPatterns(*logPaths, *logsFile)
 	if len(patterns) == 0 {
 		log.Fatal("collector: no log paths specified; use --logs or --logs-file")
 	}
-
-	paths := expandGlobs(patterns)
-	if len(paths) == 0 {
-		log.Fatal("collector: no log files matched the specified patterns")
-	}
-	log.Printf("collector: tailing %d file(s)", len(paths))
+	log.Printf("collector: watching %d pattern(s), rescan every %s", len(patterns), *scanInterval)

 	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
 	defer stop()
@@ -45,7 +42,7 @@ func main() {
 	store := NewStore(*source)
 	go store.Run(ch)

-	tailer := NewMultiTailer(paths, *v4prefix, *v6prefix, ch)
+	tailer := NewMultiTailer(patterns, *scanInterval, *v4prefix, *v6prefix, ch)
 	go tailer.Run(ctx)

 	lis, err := net.Listen("tcp", *listen)
@@ -64,7 +61,22 @@ func main() {

 	<-ctx.Done()
 	log.Printf("collector: shutting down")
-	grpcServer.GracefulStop()
+
+	// GracefulStop waits for all RPCs to finish. StreamSnapshots subscribers
+	// (e.g. the aggregator) hold a stream open indefinitely, so we give it a
+	// short window and then force-stop to avoid hanging systemctl stop/restart.
+	stopped := make(chan struct{})
+	go func() {
+		grpcServer.GracefulStop()
+		close(stopped)
+	}()
+	select {
+	case <-stopped:
+	case <-time.After(5 * time.Second):
+		log.Printf("collector: graceful stop timed out, forcing stop")
+		grpcServer.Stop()
+	}
+
 	close(ch)
 }

--- a/cmd/collector/parser.go
+++ b/cmd/collector/parser.go
@@ -6,22 +6,25 @@ import (
 	"strings"
 )

-// LogRecord holds the four dimensions extracted from a single nginx log line.
+// LogRecord holds the dimensions extracted from a single nginx log line.
 type LogRecord struct {
 	Website      string
 	ClientPrefix string
 	URI          string
 	Status       string
+	IsTor        bool
 }

 // 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
+//	$host \t $remote_addr \t $msec \t $request_method \t $request_uri \t $status \t $body_bytes_sent \t $request_time \t $is_tor
 //
+// 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.
 // Returns false for lines with fewer than 8 fields.
 func ParseLine(line string, v4bits, v6bits int) (LogRecord, bool) {
-	// SplitN caps allocations; we need exactly 8 fields.
-	fields := strings.SplitN(line, "\t", 8)
+	// SplitN caps allocations; we need up to 9 fields.
+	fields := strings.SplitN(line, "\t", 9)
 	if len(fields) < 8 {
 		return LogRecord{}, false
 	}
@@ -36,11 +39,14 @@ func ParseLine(line string, v4bits, v6bits int) (LogRecord, bool) {
 		return LogRecord{}, false
 	}

+	isTor := len(fields) == 9 && fields[8] == "1"
+
 	return LogRecord{
 		Website:      fields[0],
 		ClientPrefix: prefix,
 		URI:          uri,
 		Status:       fields[5],
+		IsTor:        isTor,
 	}, true
 }

--- a/cmd/collector/parser_test.go
+++ b/cmd/collector/parser_test.go
@@ -72,6 +72,42 @@ func TestParseLine(t *testing.T) {
 				Status:       "429",
 			},
 		},
+		{
+			name:   "is_tor=1 sets IsTor true",
+			line:   "tor.example.com\t1.2.3.4\t0\tGET\t/\t200\t0\t0.001\t1",
+			wantOK: true,
+			want: LogRecord{
+				Website:      "tor.example.com",
+				ClientPrefix: "1.2.3.0/24",
+				URI:          "/",
+				Status:       "200",
+				IsTor:        true,
+			},
+		},
+		{
+			name:   "is_tor=0 sets IsTor false",
+			line:   "normal.example.com\t1.2.3.4\t0\tGET\t/\t200\t0\t0.001\t0",
+			wantOK: true,
+			want: LogRecord{
+				Website:      "normal.example.com",
+				ClientPrefix: "1.2.3.0/24",
+				URI:          "/",
+				Status:       "200",
+				IsTor:        false,
+			},
+		},
+		{
+			name:   "missing is_tor field defaults to false (backward compat)",
+			line:   "old.example.com\t1.2.3.4\t0\tGET\t/\t200\t0\t0.001",
+			wantOK: true,
+			want: LogRecord{
+				Website:      "old.example.com",
+				ClientPrefix: "1.2.3.0/24",
+				URI:          "/",
+				Status:       "200",
+				IsTor:        false,
+			},
+		},
 	}

 	for _, tc := range tests {
--- a/cmd/collector/smoke_test.go
+++ b/cmd/collector/smoke_test.go
@@ -104,10 +104,10 @@ func TestGRPCEndToEnd(t *testing.T) {

 	// Pre-populate with known data then rotate so it's queryable
 	for i := 0; i < 500; i++ {
-		store.ingest(LogRecord{"busy.com", "1.2.3.0/24", "/api", "200"})
+		store.ingest(LogRecord{Website: "busy.com", ClientPrefix: "1.2.3.0/24", URI: "/api", Status: "200"})
 	}
 	for i := 0; i < 200; i++ {
-		store.ingest(LogRecord{"quiet.com", "5.6.7.0/24", "/", "429"})
+		store.ingest(LogRecord{Website: "quiet.com", ClientPrefix: "5.6.7.0/24", URI: "/", Status: "429"})
 	}
 	store.rotate(time.Now())

@@ -192,7 +192,7 @@ func TestGRPCEndToEnd(t *testing.T) {
 		t.Fatalf("StreamSnapshots error: %v", err)
 	}

-	store.ingest(LogRecord{"new.com", "9.9.9.0/24", "/new", "200"})
+	store.ingest(LogRecord{Website: "new.com", ClientPrefix: "9.9.9.0/24", URI: "/new", Status: "200"})
 	store.rotate(time.Now())

 	snap, err := stream.Recv()
--- a/cmd/collector/store.go
+++ b/cmd/collector/store.go
@@ -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.Tuple4]int64
+	live    map[st.Tuple5]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.Tuple4]int64, liveMapCap),
+		live:   make(map[st.Tuple5]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.Tuple4{Website: r.Website, Prefix: r.ClientPrefix, URI: r.URI, Status: r.Status}
+	key := st.Tuple5{Website: r.Website, Prefix: r.ClientPrefix, URI: r.URI, Status: r.Status, IsTor: r.IsTor}
 	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.Tuple4]int64, liveMapCap)
+	s.live = make(map[st.Tuple5]int64, liveMapCap)
 	s.liveLen = 0

 	s.broadcast(fine)
--- a/cmd/collector/store_test.go
+++ b/cmd/collector/store_test.go
@@ -15,7 +15,7 @@ func makeStore() *Store {

 func ingestN(s *Store, website, prefix, uri, status string, n int) {
 	for i := 0; i < n; i++ {
-		s.ingest(LogRecord{website, prefix, uri, status})
+		s.ingest(LogRecord{Website: website, ClientPrefix: prefix, URI: uri, Status: status})
 	}
 }

--- a/cmd/collector/tailer.go
+++ b/cmd/collector/tailer.go
@@ -28,20 +28,25 @@ type reopenMsg struct {
 // fsnotify.Watcher (one inotify instance). This scales to hundreds of files
 // without hitting the kernel limit on inotify instances per user.
 type MultiTailer struct {
-	paths   []string
-	v4bits  int
-	v6bits  int
-	ch      chan<- LogRecord
+	patterns     []string
+	scanInterval time.Duration
+	v4bits       int
+	v6bits       int
+	ch           chan<- LogRecord
 }

-func NewMultiTailer(paths []string, v4bits, v6bits int, ch chan<- LogRecord) *MultiTailer {
-	return &MultiTailer{paths: paths, v4bits: v4bits, v6bits: v6bits, ch: ch}
+func NewMultiTailer(patterns []string, scanInterval time.Duration, v4bits, v6bits int, ch chan<- LogRecord) *MultiTailer {
+	return &MultiTailer{patterns: patterns, scanInterval: scanInterval, v4bits: v4bits, v6bits: v6bits, ch: ch}
 }

 // Run tails all configured files until ctx is cancelled.
 // All files share one fsnotify.Watcher. Log rotation is handled per-file:
 // on RENAME/REMOVE the old fd is drained then a retry goroutine re-opens
 // the original path and hands it back via a channel.
+//
+// A periodic rescan re-expands the glob patterns so that files created after
+// startup are picked up automatically and files that have disappeared (and
+// are no longer matched by any pattern) are retired.
 func (mt *MultiTailer) Run(ctx context.Context) {
 	watcher, err := fsnotify.NewWatcher()
 	if err != nil {
@@ -49,21 +54,33 @@ func (mt *MultiTailer) Run(ctx context.Context) {
 	}
 	defer watcher.Close()

-	files := make(map[string]*fileState, len(mt.paths))
-	reopenCh := make(chan reopenMsg, len(mt.paths))
+	files    := make(map[string]*fileState)
+	retrying := make(map[string]struct{}) // paths currently in a retryOpen goroutine
+	reopenCh := make(chan reopenMsg, 32)

-	// Open all files and seek to EOF.
-	for _, path := range mt.paths {
+	startRetry := func(path string) {
+		if _, already := retrying[path]; already {
+			return
+		}
+		retrying[path] = struct{}{}
+		go retryOpen(ctx, path, watcher, reopenCh)
+	}
+
+	// Initial scan.
+	for _, path := range expandGlobs(mt.patterns) {
 		fs, err := openAndSeekEOF(path, watcher)
 		if err != nil {
 			log.Printf("tailer: %s not found, will retry: %v", path, err)
-			go retryOpen(ctx, path, watcher, reopenCh)
+			startRetry(path)
 			continue
 		}
 		files[path] = fs
 		log.Printf("tailer: watching %s", path)
 	}

+	ticker := time.NewTicker(mt.scanInterval)
+	defer ticker.Stop()
+
 	for {
 		select {
 		case <-ctx.Done():
@@ -76,12 +93,16 @@ func (mt *MultiTailer) Run(ctx context.Context) {
 			if !ok {
 				return
 			}
+			delete(retrying, msg.path)
 			files[msg.path] = &fileState{f: msg.f, reader: bufio.NewReader(msg.f)}
 			if err := watcher.Add(msg.path); err != nil {
 				log.Printf("tailer: watcher re-add failed for %s: %v", msg.path, err)
 			}
 			log.Printf("tailer: re-opened %s after rotation", msg.path)

+		case <-ticker.C:
+			mt.rescan(ctx, watcher, files, retrying, reopenCh, startRetry)
+
 		case event, ok := <-watcher.Events:
 			if !ok {
 				return
@@ -99,7 +120,7 @@ func (mt *MultiTailer) Run(ctx context.Context) {
 				fs.f.Close()
 				delete(files, event.Name)
 				_ = watcher.Remove(event.Name)
-				go retryOpen(ctx, event.Name, watcher, reopenCh)
+				startRetry(event.Name)
 			}

 		case err, ok := <-watcher.Errors:
@@ -111,6 +132,49 @@ func (mt *MultiTailer) Run(ctx context.Context) {
 	}
 }

+// rescan re-expands the glob patterns and reconciles against the current file
+// set: new matches are opened (or queued for retry), and files no longer
+// matched by any pattern are drained, closed, and retired.
+func (mt *MultiTailer) rescan(
+	ctx context.Context,
+	watcher *fsnotify.Watcher,
+	files map[string]*fileState,
+	retrying map[string]struct{},
+	_ chan reopenMsg,
+	startRetry func(string),
+) {
+	current := make(map[string]struct{})
+	for _, path := range expandGlobs(mt.patterns) {
+		current[path] = struct{}{}
+		if _, inFiles := files[path]; inFiles {
+			continue
+		}
+		if _, isRetrying := retrying[path]; isRetrying {
+			continue
+		}
+		// Newly matched file — try to open it right away.
+		fs, err := openAndSeekEOF(path, watcher)
+		if err != nil {
+			startRetry(path)
+			continue
+		}
+		files[path] = fs
+		log.Printf("tailer: discovered %s", path)
+	}
+
+	// Retire files that no longer match any pattern and are not being rotated
+	// (rotation is handled by the RENAME/REMOVE event path, not here).
+	for path, fs := range files {
+		if _, matched := current[path]; !matched {
+			mt.readLines(fs.reader)
+			fs.f.Close()
+			_ = watcher.Remove(path)
+			delete(files, path)
+			log.Printf("tailer: retired %s (no longer matched by any pattern)", path)
+		}
+	}
+}
+
 // openAndSeekEOF opens path, seeks to EOF, and registers it with watcher.
 func openAndSeekEOF(path string, watcher *fsnotify.Watcher) (*fileState, error) {
 	f, err := os.Open(path)
--- a/cmd/collector/tailer_test.go
+++ b/cmd/collector/tailer_test.go
@@ -28,7 +28,7 @@ func TestMultiTailerReadsLines(t *testing.T) {
 	defer f.Close()

 	ch := make(chan LogRecord, 100)
-	mt := NewMultiTailer([]string{path}, 24, 48, ch)
+	mt := NewMultiTailer([]string{path}, time.Hour, 24, 48, ch)

 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -67,7 +67,7 @@ func TestMultiTailerMultipleFiles(t *testing.T) {
 	}

 	ch := make(chan LogRecord, 200)
-	mt := NewMultiTailer(paths, 24, 48, ch)
+	mt := NewMultiTailer(paths, time.Hour, 24, 48, ch)
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	go mt.Run(ctx)
@@ -95,7 +95,7 @@ func TestMultiTailerLogRotation(t *testing.T) {
 	}

 	ch := make(chan LogRecord, 100)
-	mt := NewMultiTailer([]string{path}, 24, 48, ch)
+	mt := NewMultiTailer([]string{path}, time.Hour, 24, 48, ch)
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	go mt.Run(ctx)
--- a/cmd/frontend/filter.go
+++ b/cmd/frontend/filter.go
@@ -113,8 +113,21 @@ func applyTerm(term string, fs *filterState) error {
 			return fmt.Errorf("prefix only supports =, not %q", op)
 		}
 		fs.Prefix = value
+	case "is_tor":
+		if op != "=" && op != "!=" {
+			return fmt.Errorf("is_tor only supports = and !=, not %q", op)
+		}
+		if value != "0" && value != "1" {
+			return fmt.Errorf("is_tor value must be 0 or 1, not %q", value)
+		}
+		// Normalise: is_tor=1 and is_tor!=0 both mean "TOR only"
+		if (op == "=" && value == "1") || (op == "!=" && value == "0") {
+			fs.IsTor = "1"
+		} else {
+			fs.IsTor = "0"
+		}
 	default:
-		return fmt.Errorf("unknown field %q; valid: status, website, uri, prefix", field)
+		return fmt.Errorf("unknown field %q; valid: status, website, uri, prefix, is_tor", field)
 	}
 	return nil
 }
@@ -151,6 +164,9 @@ func FilterExprString(f filterState) string {
 	if f.Status != "" {
 		parts = append(parts, statusTermStr(f.Status))
 	}
+	if f.IsTor != "" {
+		parts = append(parts, "is_tor="+f.IsTor)
+	}
 	return strings.Join(parts, " AND ")
 }

--- a/cmd/frontend/handler.go
+++ b/cmd/frontend/handler.go
@@ -53,6 +53,7 @@ type filterState struct {
 	Status    string // expression: "200", "!=200", ">=400", etc.
 	WebsiteRe string // RE2 regex against website
 	URIRe     string // RE2 regex against request URI
+	IsTor     string // "", "1" (TOR only), "0" (non-TOR only)
 }

 // QueryParams holds all parsed URL parameters for one page request.
@@ -77,6 +78,7 @@ type PageData struct {
 	Windows        []Tab
 	GroupBys       []Tab
 	Targets        []Tab // source/target picker; empty when only one target available
+	TorTabs        []Tab // all / tor / no-tor toggle
 	RefreshSecs    int
 	Error          string
 	FilterExpr     string // current filter serialised to mini-language for the input box
@@ -156,12 +158,13 @@ func (h *Handler) parseParams(r *http.Request) QueryParams {
 			Status:    q.Get("f_status"),
 			WebsiteRe: q.Get("f_website_re"),
 			URIRe:     q.Get("f_uri_re"),
+			IsTor:     q.Get("f_is_tor"),
 		},
 	}
 }

 func buildFilter(f filterState) *pb.Filter {
-	if f.Website == "" && f.Prefix == "" && f.URI == "" && f.Status == "" && f.WebsiteRe == "" && f.URIRe == "" {
+	if f.Website == "" && f.Prefix == "" && f.URI == "" && f.Status == "" && f.WebsiteRe == "" && f.URIRe == "" && f.IsTor == "" {
 		return nil
 	}
 	out := &pb.Filter{}
@@ -186,6 +189,12 @@ func buildFilter(f filterState) *pb.Filter {
 	if f.URIRe != "" {
 		out.UriRegex = &f.URIRe
 	}
+	switch f.IsTor {
+	case "1":
+		out.Tor = pb.TorFilter_TOR_YES
+	case "0":
+		out.Tor = pb.TorFilter_TOR_NO
+	}
 	return out
 }

@@ -214,6 +223,9 @@ func (p QueryParams) toValues() url.Values {
 	if p.Filter.URIRe != "" {
 		v.Set("f_uri_re", p.Filter.URIRe)
 	}
+	if p.Filter.IsTor != "" {
+		v.Set("f_is_tor", p.Filter.IsTor)
+	}
 	return v
 }

@@ -314,6 +326,18 @@ func buildCrumbs(p QueryParams) []Crumb {
 			RemoveURL: p.buildURL(map[string]string{"f_uri_re": ""}),
 		})
 	}
+	switch p.Filter.IsTor {
+	case "1":
+		crumbs = append(crumbs, Crumb{
+			Text:      "is_tor=1 (TOR only)",
+			RemoveURL: p.buildURL(map[string]string{"f_is_tor": ""}),
+		})
+	case "0":
+		crumbs = append(crumbs, Crumb{
+			Text:      "is_tor=0 (no TOR)",
+			RemoveURL: p.buildURL(map[string]string{"f_is_tor": ""}),
+		})
+	}
 	return crumbs
 }

@@ -341,6 +365,23 @@ func buildGroupByTabs(p QueryParams) []Tab {
 	return tabs
 }

+func buildTorTabs(p QueryParams) []Tab {
+	specs := []struct{ val, label string }{
+		{"", "all"},
+		{"1", "tor"},
+		{"0", "no tor"},
+	}
+	tabs := make([]Tab, len(specs))
+	for i, s := range specs {
+		tabs[i] = Tab{
+			Label:  s.label,
+			URL:    p.buildURL(map[string]string{"f_is_tor": s.val}),
+			Active: p.Filter.IsTor == s.val,
+		}
+	}
+	return tabs
+}
+
 // buildTargetTabs builds the source/target picker tabs from a ListTargets response.
 // Returns nil (hide picker) when only one endpoint is reachable.
 func (h *Handler) buildTargetTabs(p QueryParams, lt *pb.ListTargetsResponse) []Tab {
@@ -502,6 +543,7 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		Breadcrumbs:    buildCrumbs(params),
 		Windows:        buildWindowTabs(params),
 		GroupBys:       buildGroupByTabs(params),
+		TorTabs:        buildTorTabs(params),
 		Targets:        h.buildTargetTabs(params, lt),
 		RefreshSecs:    h.refreshSecs,
 		FilterExpr:     filterExprInput,
@@ -524,6 +566,7 @@ func (h *Handler) errorPage(params QueryParams, msg string) PageData {
 		Params:         params,
 		Windows:        buildWindowTabs(params),
 		GroupBys:       buildGroupByTabs(params),
+		TorTabs:        buildTorTabs(params),
 		Breadcrumbs:    buildCrumbs(params),
 		RefreshSecs:    h.refreshSecs,
 		Error:          msg,
--- a/cmd/frontend/templates/base.html
+++ b/cmd/frontend/templates/base.html
@@ -35,6 +35,7 @@ a:hover { text-decoration: underline; }
 .nodata { color: #999; margin: 2em 0; font-style: italic; }
 footer { margin-top: 2em; padding-top: 0.6em; border-top: 1px solid #e0e0e0; font-size: 0.8em; color: #999; }
 .tabs-targets { margin-top: -0.4em; }
+.tabs-tor { margin-top: -0.4em; }
 .tabs-label { font-size: 0.85em; color: #888; margin-right: 0.2em; align-self: center; }
 .filter-form { display: flex; gap: 0.4em; align-items: center; margin-bottom: 0.7em; }
 .filter-input { flex: 1; font-family: monospace; font-size: 13px; padding: 0.25em 0.5em; border: 1px solid #aaa; }
--- a/cmd/frontend/templates/index.html
+++ b/cmd/frontend/templates/index.html
@@ -20,12 +20,19 @@
 {{- end}}
 </div>{{end}}

+<div class="tabs tabs-tor">
+  <span class="tabs-label">tor:</span>
+{{- range .TorTabs}}
+  <a href="{{.URL}}"{{if .Active}} class="active"{{end}}>{{.Label}}</a>
+{{- end}}
+</div>
+
 <form class="filter-form" method="get" action="/">
  <input type="hidden" name="target" value="{{.Params.Target}}">
  <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/">
+  <input class="filter-input" type="text" name="q" value="{{.FilterExpr}}" placeholder="status>=400 AND website~=gouda.* AND uri~=^/api/ AND is_tor=1">
  <button type="submit">filter</button>
  {{- if .FilterExpr}} <a class="clear" href="{{.ClearFilterURL}}">× clear</a>{{end}}
 </form>
--- a/docs/USERGUIDE.md
+++ b/docs/USERGUIDE.md
@@ -27,7 +27,7 @@ Add the `logtail` log format to your `nginx.conf` and apply it to each `server`

 ```nginx
 http {
-    log_format logtail '$host\t$remote_addr\t$msec\t$request_method\t$request_uri\t$status\t$body_bytes_sent\t$request_time';
+    log_format logtail '$host\t$remote_addr\t$msec\t$request_method\t$request_uri\t$status\t$body_bytes_sent\t$request_time\t$is_tor';

    server {
        access_log /var/log/nginx/access.log logtail;
@@ -38,7 +38,10 @@ http {
 ```

 The format is tab-separated with fixed field positions. Query strings are stripped from the URI
-by the collector at ingest time — only the path is tracked.
+by the collector at ingest time — only the path is tracked. `$is_tor` must be set to `1` when
+the client IP is a TOR exit node and `0` otherwise (this is typically populated by a custom nginx
+variable or a Lua script that checks the IP against a TOR exit list). The field is optional for
+backward compatibility — log lines without it are accepted and treated as `is_tor=0`.

 ---

@@ -64,14 +67,15 @@ windows, and exposes a gRPC interface for the aggregator (and directly for the C

 ### Flags

-| Flag           | Default      | Description                                               |
-|----------------|--------------|-----------------------------------------------------------|
-| `--listen`     | `:9090`      | gRPC listen address                                       |
-| `--logs`       | —            | Comma-separated log file paths or glob patterns           |
-| `--logs-file`  | —            | File containing one log path/glob per line                |
-| `--source`     | hostname     | Name for this collector in query responses                |
-| `--v4prefix`   | `24`         | IPv4 prefix length for client bucketing (e.g. /24 → /23) |
-| `--v6prefix`   | `48`         | IPv6 prefix length for client bucketing                   |
+| Flag              | Default      | Description                                               |
+|-------------------|--------------|-----------------------------------------------------------|
+| `--listen`        | `:9090`      | gRPC listen address                                       |
+| `--logs`          | —            | Comma-separated log file paths or glob patterns           |
+| `--logs-file`     | —            | File containing one log path/glob per line                |
+| `--source`        | hostname     | Name for this collector in query responses                |
+| `--v4prefix`      | `24`         | IPv4 prefix length for client bucketing (e.g. /24 → /23) |
+| `--v6prefix`      | `48`         | IPv6 prefix length for client bucketing                   |
+| `--scan-interval` | `10s`        | How often to rescan glob patterns for new/removed files   |

 At least one of `--logs` or `--logs-file` is required.

@@ -124,7 +128,7 @@ The collector is designed to stay well under 1 GB:
 | Coarse ring (288 × 5-min)   | 288 × 5 000 | ~268 MB     |
 | **Total**                   |             | **~845 MB** |

-When the live map reaches 100 000 distinct 4-tuples, new keys are dropped for the rest of that
+When the live map reaches 100 000 distinct 5-tuples, new keys are dropped for the rest of that
 minute. Existing keys continue to accumulate counts. The cap resets at each minute rotation.

 ### Time windows
@@ -284,6 +288,10 @@ Supported fields and operators:
 | `website` | `=` `~=`            | `website~=gouda.*`         |
 | `uri`     | `=` `~=`            | `uri~=^/api/`              |
 | `prefix`  | `=`                 | `prefix=1.2.3.0/24`        |
+| `is_tor`  | `=` `!=`            | `is_tor=1`, `is_tor!=0`    |
+
+`is_tor=1` and `is_tor!=0` are equivalent (TOR traffic only). `is_tor=0` and `is_tor!=1` are
+equivalent (non-TOR traffic only).

 `~=` means RE2 regex match. Values with spaces or quotes may be wrapped in double or single
 quotes: `uri~="^/search\?q="`.
@@ -303,8 +311,8 @@ accept RE2 regular expressions. The breadcrumb strip shows them as `website~=gou
 `uri~=^/api/` with the usual `×` remove link.

 **URL sharing** — all filter state is in the URL query string (`w`, `by`, `f_website`,
-`f_prefix`, `f_uri`, `f_status`, `f_website_re`, `f_uri_re`, `n`). Copy the URL to share an
-exact view with another operator, or bookmark a recurring query.
+`f_prefix`, `f_uri`, `f_status`, `f_website_re`, `f_uri_re`, `f_is_tor`, `n`). Copy the URL to
+share an exact view with another operator, or bookmark a recurring query.

 **JSON output** — append `&raw=1` to any URL to receive the TopN result as JSON instead of
 HTML. Useful for scripting without the CLI binary:
@@ -359,6 +367,7 @@ logtail-cli targets [flags]   list targets known to the queried endpoint
 | `--status`    | —                | Filter: HTTP status expression (`200`, `!=200`, `>=400`, `<500`, …) |
 | `--website-re`| —                | Filter: RE2 regex against website                        |
 | `--uri-re`    | —                | Filter: RE2 regex against request URI                    |
+| `--is-tor`    | —                | Filter: `1` or `!=0` = TOR only; `0` or `!=1` = non-TOR only |

 ### `topn` flags

@@ -455,6 +464,12 @@ logtail-cli topn --target agg:9091 --window 5m --website-re 'gouda.*'
 # Filter by URI regex: all /api/ paths
 logtail-cli topn --target agg:9091 --window 5m --group-by uri --uri-re '^/api/'

+# Show only TOR traffic — which websites are TOR clients hitting?
+logtail-cli topn --target agg:9091 --window 5m --is-tor 1
+
+# Show non-TOR traffic only — exclude exit nodes from the view
+logtail-cli topn --target agg:9091 --window 5m --is-tor 0
+
 # Compare two collectors side by side in one command
 logtail-cli topn --target nginx1:9090,nginx2:9090 --window 5m

--- a/internal/store/store.go
+++ b/internal/store/store.go
@@ -20,12 +20,13 @@ const (
 	CoarseEvery    = 5       // fine ticks between coarse writes
 )

-// Tuple4 is the four-dimensional aggregation key.
-type Tuple4 struct {
+// Tuple5 is the aggregation key (website, prefix, URI, status, is_tor).
+type Tuple5 struct {
 	Website string
 	Prefix  string
 	URI     string
 	Status  string
+	IsTor   bool
 }

 // Entry is a labelled count used in snapshots and query results.
@@ -73,21 +74,29 @@ func BucketsForWindow(window pb.Window, fine, coarse RingView, fineFilled, coars
 	}
 }

-// --- label encoding: "website\x00prefix\x00uri\x00status" ---
+// --- label encoding: "website\x00prefix\x00uri\x00status\x00is_tor" ---

-// EncodeTuple encodes a Tuple4 as a NUL-separated string suitable for use
+// EncodeTuple encodes a Tuple5 as a NUL-separated string suitable for use
 // as a map key in snapshots.
-func EncodeTuple(t Tuple4) string {
-	return t.Website + "\x00" + t.Prefix + "\x00" + t.URI + "\x00" + t.Status
+func EncodeTuple(t Tuple5) string {
+	tor := "0"
+	if t.IsTor {
+		tor = "1"
+	}
+	return t.Website + "\x00" + t.Prefix + "\x00" + t.URI + "\x00" + t.Status + "\x00" + tor
 }

-// LabelTuple decodes a NUL-separated snapshot label back into a Tuple4.
-func LabelTuple(label string) Tuple4 {
-	parts := splitN(label, '\x00', 4)
-	if len(parts) != 4 {
-		return Tuple4{}
+// LabelTuple decodes a NUL-separated snapshot label back into a Tuple5.
+func LabelTuple(label string) Tuple5 {
+	parts := splitN(label, '\x00', 5)
+	if len(parts) < 4 {
+		return Tuple5{}
 	}
-	return Tuple4{parts[0], parts[1], parts[2], parts[3]}
+	t := Tuple5{Website: parts[0], Prefix: parts[1], URI: parts[2], Status: parts[3]}
+	if len(parts) == 5 {
+		t.IsTor = parts[4] == "1"
+	}
+	return t
 }

 func splitN(s string, sep byte, n int) []string {
@@ -150,7 +159,7 @@ func CompileFilter(f *pb.Filter) *CompiledFilter {

 // MatchesFilter returns true if t satisfies all constraints in f.
 // A nil filter matches everything.
-func MatchesFilter(t Tuple4, f *CompiledFilter) bool {
+func MatchesFilter(t Tuple5, f *CompiledFilter) bool {
 	if f == nil || f.Proto == nil {
 		return true
 	}
@@ -180,6 +189,16 @@ func MatchesFilter(t Tuple4, f *CompiledFilter) bool {
 	if p.HttpResponse != nil && !matchesStatusOp(t.Status, p.GetHttpResponse(), p.StatusOp) {
 		return false
 	}
+	switch p.Tor {
+	case pb.TorFilter_TOR_YES:
+		if !t.IsTor {
+			return false
+		}
+	case pb.TorFilter_TOR_NO:
+		if t.IsTor {
+			return false
+		}
+	}
 	return true
 }

@@ -210,7 +229,7 @@ func matchesStatusOp(statusStr string, want int32, op pb.StatusOp) bool {
 }

 // DimensionLabel returns the string value of t for the given group-by dimension.
-func DimensionLabel(t Tuple4, g pb.GroupBy) string {
+func DimensionLabel(t Tuple5, g pb.GroupBy) string {
 	switch g {
 	case pb.GroupBy_WEBSITE:
 		return t.Website
@@ -299,9 +318,9 @@ func TopKFromMap(m map[string]int64, k int) []Entry {
 	return result
 }

-// TopKFromTupleMap encodes a Tuple4 map and returns the top-k as a Snapshot.
+// TopKFromTupleMap encodes a Tuple5 map and returns the top-k as a Snapshot.
 // Used by the collector to snapshot its live map.
-func TopKFromTupleMap(m map[Tuple4]int64, k int, ts time.Time) Snapshot {
+func TopKFromTupleMap(m map[Tuple5]int64, k int, ts time.Time) Snapshot {
 	flat := make(map[string]int64, len(m))
 	for t, c := range m {
 		flat[EncodeTuple(t)] = c
--- a/internal/store/store_test.go
+++ b/internal/store/store_test.go
@@ -83,10 +83,10 @@ func compiledEQ(status int32) *CompiledFilter {
 }

 func TestMatchesFilterNil(t *testing.T) {
-	if !MatchesFilter(Tuple4{Website: "x"}, nil) {
+	if !MatchesFilter(Tuple5{Website: "x"}, nil) {
 		t.Fatal("nil filter should match everything")
 	}
-	if !MatchesFilter(Tuple4{Website: "x"}, &CompiledFilter{}) {
+	if !MatchesFilter(Tuple5{Website: "x"}, &CompiledFilter{}) {
 		t.Fatal("empty compiled filter should match everything")
 	}
 }
@@ -94,10 +94,10 @@ func TestMatchesFilterNil(t *testing.T) {
 func TestMatchesFilterExactWebsite(t *testing.T) {
 	w := "example.com"
 	cf := CompileFilter(&pb.Filter{Website: &w})
-	if !MatchesFilter(Tuple4{Website: "example.com"}, cf) {
+	if !MatchesFilter(Tuple5{Website: "example.com"}, cf) {
 		t.Fatal("expected match")
 	}
-	if MatchesFilter(Tuple4{Website: "other.com"}, cf) {
+	if MatchesFilter(Tuple5{Website: "other.com"}, cf) {
 		t.Fatal("expected no match")
 	}
 }
@@ -105,10 +105,10 @@ func TestMatchesFilterExactWebsite(t *testing.T) {
 func TestMatchesFilterWebsiteRegex(t *testing.T) {
 	re := "gouda.*"
 	cf := CompileFilter(&pb.Filter{WebsiteRegex: &re})
-	if !MatchesFilter(Tuple4{Website: "gouda.example.com"}, cf) {
+	if !MatchesFilter(Tuple5{Website: "gouda.example.com"}, cf) {
 		t.Fatal("expected match")
 	}
-	if MatchesFilter(Tuple4{Website: "edam.example.com"}, cf) {
+	if MatchesFilter(Tuple5{Website: "edam.example.com"}, cf) {
 		t.Fatal("expected no match")
 	}
 }
@@ -116,10 +116,10 @@ func TestMatchesFilterWebsiteRegex(t *testing.T) {
 func TestMatchesFilterURIRegex(t *testing.T) {
 	re := "^/api/.*"
 	cf := CompileFilter(&pb.Filter{UriRegex: &re})
-	if !MatchesFilter(Tuple4{URI: "/api/users"}, cf) {
+	if !MatchesFilter(Tuple5{URI: "/api/users"}, cf) {
 		t.Fatal("expected match")
 	}
-	if MatchesFilter(Tuple4{URI: "/health"}, cf) {
+	if MatchesFilter(Tuple5{URI: "/health"}, cf) {
 		t.Fatal("expected no match")
 	}
 }
@@ -127,17 +127,17 @@ func TestMatchesFilterURIRegex(t *testing.T) {
 func TestMatchesFilterInvalidRegexMatchesNothing(t *testing.T) {
 	re := "[invalid"
 	cf := CompileFilter(&pb.Filter{WebsiteRegex: &re})
-	if MatchesFilter(Tuple4{Website: "anything"}, cf) {
+	if MatchesFilter(Tuple5{Website: "anything"}, cf) {
 		t.Fatal("invalid regex should match nothing")
 	}
 }

 func TestMatchesFilterStatusEQ(t *testing.T) {
 	cf := compiledEQ(200)
-	if !MatchesFilter(Tuple4{Status: "200"}, cf) {
+	if !MatchesFilter(Tuple5{Status: "200"}, cf) {
 		t.Fatal("expected match")
 	}
-	if MatchesFilter(Tuple4{Status: "404"}, cf) {
+	if MatchesFilter(Tuple5{Status: "404"}, cf) {
 		t.Fatal("expected no match")
 	}
 }
@@ -145,10 +145,10 @@ func TestMatchesFilterStatusEQ(t *testing.T) {
 func TestMatchesFilterStatusNE(t *testing.T) {
 	v := int32(200)
 	cf := CompileFilter(&pb.Filter{HttpResponse: &v, StatusOp: pb.StatusOp_NE})
-	if MatchesFilter(Tuple4{Status: "200"}, cf) {
+	if MatchesFilter(Tuple5{Status: "200"}, cf) {
 		t.Fatal("expected no match for 200 != 200")
 	}
-	if !MatchesFilter(Tuple4{Status: "404"}, cf) {
+	if !MatchesFilter(Tuple5{Status: "404"}, cf) {
 		t.Fatal("expected match for 404 != 200")
 	}
 }
@@ -156,13 +156,13 @@ func TestMatchesFilterStatusNE(t *testing.T) {
 func TestMatchesFilterStatusGE(t *testing.T) {
 	v := int32(400)
 	cf := CompileFilter(&pb.Filter{HttpResponse: &v, StatusOp: pb.StatusOp_GE})
-	if !MatchesFilter(Tuple4{Status: "400"}, cf) {
+	if !MatchesFilter(Tuple5{Status: "400"}, cf) {
 		t.Fatal("expected match: 400 >= 400")
 	}
-	if !MatchesFilter(Tuple4{Status: "500"}, cf) {
+	if !MatchesFilter(Tuple5{Status: "500"}, cf) {
 		t.Fatal("expected match: 500 >= 400")
 	}
-	if MatchesFilter(Tuple4{Status: "200"}, cf) {
+	if MatchesFilter(Tuple5{Status: "200"}, cf) {
 		t.Fatal("expected no match: 200 >= 400")
 	}
 }
@@ -170,17 +170,17 @@ func TestMatchesFilterStatusGE(t *testing.T) {
 func TestMatchesFilterStatusLT(t *testing.T) {
 	v := int32(400)
 	cf := CompileFilter(&pb.Filter{HttpResponse: &v, StatusOp: pb.StatusOp_LT})
-	if !MatchesFilter(Tuple4{Status: "200"}, cf) {
+	if !MatchesFilter(Tuple5{Status: "200"}, cf) {
 		t.Fatal("expected match: 200 < 400")
 	}
-	if MatchesFilter(Tuple4{Status: "400"}, cf) {
+	if MatchesFilter(Tuple5{Status: "400"}, cf) {
 		t.Fatal("expected no match: 400 < 400")
 	}
 }

 func TestMatchesFilterStatusNonNumeric(t *testing.T) {
 	cf := compiledEQ(200)
-	if MatchesFilter(Tuple4{Status: "ok"}, cf) {
+	if MatchesFilter(Tuple5{Status: "ok"}, cf) {
 		t.Fatal("non-numeric status should not match")
 	}
 }
@@ -193,13 +193,67 @@ func TestMatchesFilterCombined(t *testing.T) {
 		HttpResponse: &v,
 		StatusOp:     pb.StatusOp_EQ,
 	})
-	if !MatchesFilter(Tuple4{Website: "example.com", Status: "200"}, cf) {
+	if !MatchesFilter(Tuple5{Website: "example.com", Status: "200"}, cf) {
 		t.Fatal("expected match")
 	}
-	if MatchesFilter(Tuple4{Website: "other.com", Status: "200"}, cf) {
+	if MatchesFilter(Tuple5{Website: "other.com", Status: "200"}, cf) {
 		t.Fatal("expected no match: wrong website")
 	}
-	if MatchesFilter(Tuple4{Website: "example.com", Status: "404"}, cf) {
+	if MatchesFilter(Tuple5{Website: "example.com", Status: "404"}, cf) {
 		t.Fatal("expected no match: wrong status")
 	}
 }
+
+// --- IsTor label encoding and filtering ---
+
+func TestEncodeLabelTupleRoundtripWithTor(t *testing.T) {
+	for _, isTor := range []bool{false, true} {
+		orig := Tuple5{Website: "a.com", Prefix: "1.2.3.0/24", URI: "/x", Status: "200", IsTor: isTor}
+		got := LabelTuple(EncodeTuple(orig))
+		if got != orig {
+			t.Errorf("roundtrip mismatch: got %+v, want %+v", got, orig)
+		}
+	}
+}
+
+func TestLabelTupleBackwardCompat(t *testing.T) {
+	// Old 4-field label (no is_tor field) should decode with IsTor=false.
+	label := "a.com\x001.2.3.0/24\x00/x\x00200"
+	got := LabelTuple(label)
+	if got.IsTor {
+		t.Errorf("expected IsTor=false for old label, got true")
+	}
+	if got.Website != "a.com" || got.Status != "200" {
+		t.Errorf("unexpected tuple: %+v", got)
+	}
+}
+
+func TestMatchesFilterTorYes(t *testing.T) {
+	cf := CompileFilter(&pb.Filter{Tor: pb.TorFilter_TOR_YES})
+	if !MatchesFilter(Tuple5{IsTor: true}, cf) {
+		t.Fatal("TOR_YES should match TOR tuple")
+	}
+	if MatchesFilter(Tuple5{IsTor: false}, cf) {
+		t.Fatal("TOR_YES should not match non-TOR tuple")
+	}
+}
+
+func TestMatchesFilterTorNo(t *testing.T) {
+	cf := CompileFilter(&pb.Filter{Tor: pb.TorFilter_TOR_NO})
+	if !MatchesFilter(Tuple5{IsTor: false}, cf) {
+		t.Fatal("TOR_NO should match non-TOR tuple")
+	}
+	if MatchesFilter(Tuple5{IsTor: true}, cf) {
+		t.Fatal("TOR_NO should not match TOR tuple")
+	}
+}
+
+func TestMatchesFilterTorAny(t *testing.T) {
+	cf := CompileFilter(&pb.Filter{Tor: pb.TorFilter_TOR_ANY})
+	if !MatchesFilter(Tuple5{IsTor: true}, cf) {
+		t.Fatal("TOR_ANY should match TOR tuple")
+	}
+	if !MatchesFilter(Tuple5{IsTor: false}, cf) {
+		t.Fatal("TOR_ANY should match non-TOR tuple")
+	}
+}
--- a/proto/logtail.pb.go
+++ b/proto/logtail.pb.go
--- a/proto/logtail.proto
+++ b/proto/logtail.proto
@@ -4,6 +4,14 @@ package logtail;

 option go_package = "git.ipng.ch/ipng/nginx-logtail/proto/logtailpb";

+// TorFilter restricts results by whether the client is a TOR exit node.
+// TOR_ANY (0) is the default and means no filtering.
+enum TorFilter {
+  TOR_ANY = 0;  // no filter
+  TOR_YES = 1;  // only TOR traffic  (is_tor=1)
+  TOR_NO  = 2;  // only non-TOR traffic (is_tor=0)
+}
+
 // StatusOp is the comparison operator applied to http_response in a Filter.
 // Defaults to EQ (exact match) for backward compatibility.
 enum StatusOp {
@@ -25,6 +33,7 @@ message Filter {
  StatusOp        status_op          = 5;  // operator for http_response; ignored when unset
  optional string website_regex      = 6;  // RE2 regex matched against website
  optional string uri_regex          = 7;  // RE2 regex matched against http_request_uri
+  TorFilter       tor                = 8;  // restrict to TOR / non-TOR clients
 }

 enum GroupBy {
--- a/proto/logtail_grpc.pb.go
+++ b/proto/logtail_grpc.pb.go
@@ -0,0 +1,239 @@
+// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
+// versions:
+// - protoc-gen-go-grpc v1.6.1
+// - protoc             v3.21.12
+// source: proto/logtail.proto
+
+package logtailpb
+
+import (
+	context "context"
+	grpc "google.golang.org/grpc"
+	codes "google.golang.org/grpc/codes"
+	status "google.golang.org/grpc/status"
+)
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the grpc package it is being compiled against.
+// Requires gRPC-Go v1.64.0 or later.
+const _ = grpc.SupportPackageIsVersion9
+
+const (
+	LogtailService_TopN_FullMethodName            = "/logtail.LogtailService/TopN"
+	LogtailService_Trend_FullMethodName           = "/logtail.LogtailService/Trend"
+	LogtailService_StreamSnapshots_FullMethodName = "/logtail.LogtailService/StreamSnapshots"
+	LogtailService_ListTargets_FullMethodName     = "/logtail.LogtailService/ListTargets"
+)
+
+// LogtailServiceClient is the client API for LogtailService service.
+//
+// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
+type LogtailServiceClient interface {
+	TopN(ctx context.Context, in *TopNRequest, opts ...grpc.CallOption) (*TopNResponse, error)
+	Trend(ctx context.Context, in *TrendRequest, opts ...grpc.CallOption) (*TrendResponse, error)
+	StreamSnapshots(ctx context.Context, in *SnapshotRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[Snapshot], error)
+	ListTargets(ctx context.Context, in *ListTargetsRequest, opts ...grpc.CallOption) (*ListTargetsResponse, error)
+}
+
+type logtailServiceClient struct {
+	cc grpc.ClientConnInterface
+}
+
+func NewLogtailServiceClient(cc grpc.ClientConnInterface) LogtailServiceClient {
+	return &logtailServiceClient{cc}
+}
+
+func (c *logtailServiceClient) TopN(ctx context.Context, in *TopNRequest, opts ...grpc.CallOption) (*TopNResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(TopNResponse)
+	err := c.cc.Invoke(ctx, LogtailService_TopN_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *logtailServiceClient) Trend(ctx context.Context, in *TrendRequest, opts ...grpc.CallOption) (*TrendResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(TrendResponse)
+	err := c.cc.Invoke(ctx, LogtailService_Trend_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *logtailServiceClient) StreamSnapshots(ctx context.Context, in *SnapshotRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[Snapshot], error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	stream, err := c.cc.NewStream(ctx, &LogtailService_ServiceDesc.Streams[0], LogtailService_StreamSnapshots_FullMethodName, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	x := &grpc.GenericClientStream[SnapshotRequest, Snapshot]{ClientStream: stream}
+	if err := x.ClientStream.SendMsg(in); err != nil {
+		return nil, err
+	}
+	if err := x.ClientStream.CloseSend(); err != nil {
+		return nil, err
+	}
+	return x, nil
+}
+
+// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
+type LogtailService_StreamSnapshotsClient = grpc.ServerStreamingClient[Snapshot]
+
+func (c *logtailServiceClient) ListTargets(ctx context.Context, in *ListTargetsRequest, opts ...grpc.CallOption) (*ListTargetsResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(ListTargetsResponse)
+	err := c.cc.Invoke(ctx, LogtailService_ListTargets_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+// LogtailServiceServer is the server API for LogtailService service.
+// All implementations must embed UnimplementedLogtailServiceServer
+// for forward compatibility.
+type LogtailServiceServer interface {
+	TopN(context.Context, *TopNRequest) (*TopNResponse, error)
+	Trend(context.Context, *TrendRequest) (*TrendResponse, error)
+	StreamSnapshots(*SnapshotRequest, grpc.ServerStreamingServer[Snapshot]) error
+	ListTargets(context.Context, *ListTargetsRequest) (*ListTargetsResponse, error)
+	mustEmbedUnimplementedLogtailServiceServer()
+}
+
+// UnimplementedLogtailServiceServer must be embedded to have
+// forward compatible implementations.
+//
+// NOTE: this should be embedded by value instead of pointer to avoid a nil
+// pointer dereference when methods are called.
+type UnimplementedLogtailServiceServer struct{}
+
+func (UnimplementedLogtailServiceServer) TopN(context.Context, *TopNRequest) (*TopNResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "method TopN not implemented")
+}
+func (UnimplementedLogtailServiceServer) Trend(context.Context, *TrendRequest) (*TrendResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "method Trend not implemented")
+}
+func (UnimplementedLogtailServiceServer) StreamSnapshots(*SnapshotRequest, grpc.ServerStreamingServer[Snapshot]) error {
+	return status.Error(codes.Unimplemented, "method StreamSnapshots not implemented")
+}
+func (UnimplementedLogtailServiceServer) ListTargets(context.Context, *ListTargetsRequest) (*ListTargetsResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "method ListTargets not implemented")
+}
+func (UnimplementedLogtailServiceServer) mustEmbedUnimplementedLogtailServiceServer() {}
+func (UnimplementedLogtailServiceServer) testEmbeddedByValue()                        {}
+
+// UnsafeLogtailServiceServer may be embedded to opt out of forward compatibility for this service.
+// Use of this interface is not recommended, as added methods to LogtailServiceServer will
+// result in compilation errors.
+type UnsafeLogtailServiceServer interface {
+	mustEmbedUnimplementedLogtailServiceServer()
+}
+
+func RegisterLogtailServiceServer(s grpc.ServiceRegistrar, srv LogtailServiceServer) {
+	// If the following call panics, it indicates UnimplementedLogtailServiceServer was
+	// embedded by pointer and is nil.  This will cause panics if an
+	// unimplemented method is ever invoked, so we test this at initialization
+	// time to prevent it from happening at runtime later due to I/O.
+	if t, ok := srv.(interface{ testEmbeddedByValue() }); ok {
+		t.testEmbeddedByValue()
+	}
+	s.RegisterService(&LogtailService_ServiceDesc, srv)
+}
+
+func _LogtailService_TopN_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(TopNRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(LogtailServiceServer).TopN(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: LogtailService_TopN_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(LogtailServiceServer).TopN(ctx, req.(*TopNRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _LogtailService_Trend_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(TrendRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(LogtailServiceServer).Trend(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: LogtailService_Trend_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(LogtailServiceServer).Trend(ctx, req.(*TrendRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _LogtailService_StreamSnapshots_Handler(srv interface{}, stream grpc.ServerStream) error {
+	m := new(SnapshotRequest)
+	if err := stream.RecvMsg(m); err != nil {
+		return err
+	}
+	return srv.(LogtailServiceServer).StreamSnapshots(m, &grpc.GenericServerStream[SnapshotRequest, Snapshot]{ServerStream: stream})
+}
+
+// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
+type LogtailService_StreamSnapshotsServer = grpc.ServerStreamingServer[Snapshot]
+
+func _LogtailService_ListTargets_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(ListTargetsRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(LogtailServiceServer).ListTargets(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: LogtailService_ListTargets_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(LogtailServiceServer).ListTargets(ctx, req.(*ListTargetsRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+// LogtailService_ServiceDesc is the grpc.ServiceDesc for LogtailService service.
+// It's only intended for direct use with grpc.RegisterService,
+// and not to be introspected or modified (even as a copy)
+var LogtailService_ServiceDesc = grpc.ServiceDesc{
+	ServiceName: "logtail.LogtailService",
+	HandlerType: (*LogtailServiceServer)(nil),
+	Methods: []grpc.MethodDesc{
+		{
+			MethodName: "TopN",
+			Handler:    _LogtailService_TopN_Handler,
+		},
+		{
+			MethodName: "Trend",
+			Handler:    _LogtailService_Trend_Handler,
+		},
+		{
+			MethodName: "ListTargets",
+			Handler:    _LogtailService_ListTargets_Handler,
+		},
+	},
+	Streams: []grpc.StreamDesc{
+		{
+			StreamName:    "StreamSnapshots",
+			Handler:       _LogtailService_StreamSnapshots_Handler,
+			ServerStreams: true,
+		},
+	},
+	Metadata: "proto/logtail.proto",
+}
--- a/proto/logtailpb/logtail.pb.go
+++ b/proto/logtailpb/logtail.pb.go
@@ -2,7 +2,7 @@
 // versions:
 // 	protoc-gen-go v1.36.11
 // 	protoc        v3.21.12
-// source: logtail.proto
+// source: proto/logtail.proto

 package logtailpb

@@ -21,6 +21,57 @@ const (
 	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
 )

+// TorFilter restricts results by whether the client is a TOR exit node.
+// TOR_ANY (0) is the default and means no filtering.
+type TorFilter int32
+
+const (
+	TorFilter_TOR_ANY TorFilter = 0 // no filter
+	TorFilter_TOR_YES TorFilter = 1 // only TOR traffic  (is_tor=1)
+	TorFilter_TOR_NO  TorFilter = 2 // only non-TOR traffic (is_tor=0)
+)
+
+// Enum value maps for TorFilter.
+var (
+	TorFilter_name = map[int32]string{
+		0: "TOR_ANY",
+		1: "TOR_YES",
+		2: "TOR_NO",
+	}
+	TorFilter_value = map[string]int32{
+		"TOR_ANY": 0,
+		"TOR_YES": 1,
+		"TOR_NO":  2,
+	}
+)
+
+func (x TorFilter) Enum() *TorFilter {
+	p := new(TorFilter)
+	*p = x
+	return p
+}
+
+func (x TorFilter) String() string {
+	return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x))
+}
+
+func (TorFilter) Descriptor() protoreflect.EnumDescriptor {
+	return file_proto_logtail_proto_enumTypes[0].Descriptor()
+}
+
+func (TorFilter) Type() protoreflect.EnumType {
+	return &file_proto_logtail_proto_enumTypes[0]
+}
+
+func (x TorFilter) Number() protoreflect.EnumNumber {
+	return protoreflect.EnumNumber(x)
+}
+
+// Deprecated: Use TorFilter.Descriptor instead.
+func (TorFilter) EnumDescriptor() ([]byte, []int) {
+	return file_proto_logtail_proto_rawDescGZIP(), []int{0}
+}
+
 // StatusOp is the comparison operator applied to http_response in a Filter.
 // Defaults to EQ (exact match) for backward compatibility.
 type StatusOp int32
@@ -65,11 +116,11 @@ func (x StatusOp) String() string {
 }

 func (StatusOp) Descriptor() protoreflect.EnumDescriptor {
-	return file_logtail_proto_enumTypes[0].Descriptor()
+	return file_proto_logtail_proto_enumTypes[1].Descriptor()
 }

 func (StatusOp) Type() protoreflect.EnumType {
-	return &file_logtail_proto_enumTypes[0]
+	return &file_proto_logtail_proto_enumTypes[1]
 }

 func (x StatusOp) Number() protoreflect.EnumNumber {
@@ -78,7 +129,7 @@ func (x StatusOp) Number() protoreflect.EnumNumber {

 // Deprecated: Use StatusOp.Descriptor instead.
 func (StatusOp) EnumDescriptor() ([]byte, []int) {
-	return file_logtail_proto_rawDescGZIP(), []int{0}
+	return file_proto_logtail_proto_rawDescGZIP(), []int{1}
 }

 type GroupBy int32
@@ -117,11 +168,11 @@ func (x GroupBy) String() string {
 }

 func (GroupBy) Descriptor() protoreflect.EnumDescriptor {
-	return file_logtail_proto_enumTypes[1].Descriptor()
+	return file_proto_logtail_proto_enumTypes[2].Descriptor()
 }

 func (GroupBy) Type() protoreflect.EnumType {
-	return &file_logtail_proto_enumTypes[1]
+	return &file_proto_logtail_proto_enumTypes[2]
 }

 func (x GroupBy) Number() protoreflect.EnumNumber {
@@ -130,7 +181,7 @@ func (x GroupBy) Number() protoreflect.EnumNumber {

 // Deprecated: Use GroupBy.Descriptor instead.
 func (GroupBy) EnumDescriptor() ([]byte, []int) {
-	return file_logtail_proto_rawDescGZIP(), []int{1}
+	return file_proto_logtail_proto_rawDescGZIP(), []int{2}
 }

 type Window int32
@@ -175,11 +226,11 @@ func (x Window) String() string {
 }

 func (Window) Descriptor() protoreflect.EnumDescriptor {
-	return file_logtail_proto_enumTypes[2].Descriptor()
+	return file_proto_logtail_proto_enumTypes[3].Descriptor()
 }

 func (Window) Type() protoreflect.EnumType {
-	return &file_logtail_proto_enumTypes[2]
+	return &file_proto_logtail_proto_enumTypes[3]
 }

 func (x Window) Number() protoreflect.EnumNumber {
@@ -188,7 +239,7 @@ func (x Window) Number() protoreflect.EnumNumber {

 // Deprecated: Use Window.Descriptor instead.
 func (Window) EnumDescriptor() ([]byte, []int) {
-	return file_logtail_proto_rawDescGZIP(), []int{2}
+	return file_proto_logtail_proto_rawDescGZIP(), []int{3}
 }

 // Filter restricts results to entries matching all specified fields.
@@ -202,13 +253,14 @@ type Filter struct {
 	StatusOp       StatusOp               `protobuf:"varint,5,opt,name=status_op,json=statusOp,proto3,enum=logtail.StatusOp" json:"status_op,omitempty"` // operator for http_response; ignored when unset
 	WebsiteRegex   *string                `protobuf:"bytes,6,opt,name=website_regex,json=websiteRegex,proto3,oneof" json:"website_regex,omitempty"`      // RE2 regex matched against website
 	UriRegex       *string                `protobuf:"bytes,7,opt,name=uri_regex,json=uriRegex,proto3,oneof" json:"uri_regex,omitempty"`                  // RE2 regex matched against http_request_uri
+	Tor            TorFilter              `protobuf:"varint,8,opt,name=tor,proto3,enum=logtail.TorFilter" json:"tor,omitempty"`                          // restrict to TOR / non-TOR clients
 	unknownFields  protoimpl.UnknownFields
 	sizeCache      protoimpl.SizeCache
 }

 func (x *Filter) Reset() {
 	*x = Filter{}
-	mi := &file_logtail_proto_msgTypes[0]
+	mi := &file_proto_logtail_proto_msgTypes[0]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -220,7 +272,7 @@ func (x *Filter) String() string {
 func (*Filter) ProtoMessage() {}

 func (x *Filter) ProtoReflect() protoreflect.Message {
-	mi := &file_logtail_proto_msgTypes[0]
+	mi := &file_proto_logtail_proto_msgTypes[0]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -233,7 +285,7 @@ func (x *Filter) ProtoReflect() protoreflect.Message {

 // Deprecated: Use Filter.ProtoReflect.Descriptor instead.
 func (*Filter) Descriptor() ([]byte, []int) {
-	return file_logtail_proto_rawDescGZIP(), []int{0}
+	return file_proto_logtail_proto_rawDescGZIP(), []int{0}
 }

 func (x *Filter) GetWebsite() string {
@@ -285,6 +337,13 @@ func (x *Filter) GetUriRegex() string {
 	return ""
 }

+func (x *Filter) GetTor() TorFilter {
+	if x != nil {
+		return x.Tor
+	}
+	return TorFilter_TOR_ANY
+}
+
 type TopNRequest struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	Filter        *Filter                `protobuf:"bytes,1,opt,name=filter,proto3" json:"filter,omitempty"`
@@ -297,7 +356,7 @@ type TopNRequest struct {

 func (x *TopNRequest) Reset() {
 	*x = TopNRequest{}
-	mi := &file_logtail_proto_msgTypes[1]
+	mi := &file_proto_logtail_proto_msgTypes[1]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -309,7 +368,7 @@ func (x *TopNRequest) String() string {
 func (*TopNRequest) ProtoMessage() {}

 func (x *TopNRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_logtail_proto_msgTypes[1]
+	mi := &file_proto_logtail_proto_msgTypes[1]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -322,7 +381,7 @@ func (x *TopNRequest) ProtoReflect() protoreflect.Message {

 // Deprecated: Use TopNRequest.ProtoReflect.Descriptor instead.
 func (*TopNRequest) Descriptor() ([]byte, []int) {
-	return file_logtail_proto_rawDescGZIP(), []int{1}
+	return file_proto_logtail_proto_rawDescGZIP(), []int{1}
 }

 func (x *TopNRequest) GetFilter() *Filter {
@@ -363,7 +422,7 @@ type TopNEntry struct {

 func (x *TopNEntry) Reset() {
 	*x = TopNEntry{}
-	mi := &file_logtail_proto_msgTypes[2]
+	mi := &file_proto_logtail_proto_msgTypes[2]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -375,7 +434,7 @@ func (x *TopNEntry) String() string {
 func (*TopNEntry) ProtoMessage() {}

 func (x *TopNEntry) ProtoReflect() protoreflect.Message {
-	mi := &file_logtail_proto_msgTypes[2]
+	mi := &file_proto_logtail_proto_msgTypes[2]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -388,7 +447,7 @@ func (x *TopNEntry) ProtoReflect() protoreflect.Message {

 // Deprecated: Use TopNEntry.ProtoReflect.Descriptor instead.
 func (*TopNEntry) Descriptor() ([]byte, []int) {
-	return file_logtail_proto_rawDescGZIP(), []int{2}
+	return file_proto_logtail_proto_rawDescGZIP(), []int{2}
 }

 func (x *TopNEntry) GetLabel() string {
@@ -415,7 +474,7 @@ type TopNResponse struct {

 func (x *TopNResponse) Reset() {
 	*x = TopNResponse{}
-	mi := &file_logtail_proto_msgTypes[3]
+	mi := &file_proto_logtail_proto_msgTypes[3]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -427,7 +486,7 @@ func (x *TopNResponse) String() string {
 func (*TopNResponse) ProtoMessage() {}

 func (x *TopNResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_logtail_proto_msgTypes[3]
+	mi := &file_proto_logtail_proto_msgTypes[3]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -440,7 +499,7 @@ func (x *TopNResponse) ProtoReflect() protoreflect.Message {

 // Deprecated: Use TopNResponse.ProtoReflect.Descriptor instead.
 func (*TopNResponse) Descriptor() ([]byte, []int) {
-	return file_logtail_proto_rawDescGZIP(), []int{3}
+	return file_proto_logtail_proto_rawDescGZIP(), []int{3}
 }

 func (x *TopNResponse) GetEntries() []*TopNEntry {
@@ -467,7 +526,7 @@ type TrendRequest struct {

 func (x *TrendRequest) Reset() {
 	*x = TrendRequest{}
-	mi := &file_logtail_proto_msgTypes[4]
+	mi := &file_proto_logtail_proto_msgTypes[4]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -479,7 +538,7 @@ func (x *TrendRequest) String() string {
 func (*TrendRequest) ProtoMessage() {}

 func (x *TrendRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_logtail_proto_msgTypes[4]
+	mi := &file_proto_logtail_proto_msgTypes[4]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -492,7 +551,7 @@ func (x *TrendRequest) ProtoReflect() protoreflect.Message {

 // Deprecated: Use TrendRequest.ProtoReflect.Descriptor instead.
 func (*TrendRequest) Descriptor() ([]byte, []int) {
-	return file_logtail_proto_rawDescGZIP(), []int{4}
+	return file_proto_logtail_proto_rawDescGZIP(), []int{4}
 }

 func (x *TrendRequest) GetFilter() *Filter {
@@ -519,7 +578,7 @@ type TrendPoint struct {

 func (x *TrendPoint) Reset() {
 	*x = TrendPoint{}
-	mi := &file_logtail_proto_msgTypes[5]
+	mi := &file_proto_logtail_proto_msgTypes[5]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -531,7 +590,7 @@ func (x *TrendPoint) String() string {
 func (*TrendPoint) ProtoMessage() {}

 func (x *TrendPoint) ProtoReflect() protoreflect.Message {
-	mi := &file_logtail_proto_msgTypes[5]
+	mi := &file_proto_logtail_proto_msgTypes[5]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -544,7 +603,7 @@ func (x *TrendPoint) ProtoReflect() protoreflect.Message {

 // Deprecated: Use TrendPoint.ProtoReflect.Descriptor instead.
 func (*TrendPoint) Descriptor() ([]byte, []int) {
-	return file_logtail_proto_rawDescGZIP(), []int{5}
+	return file_proto_logtail_proto_rawDescGZIP(), []int{5}
 }

 func (x *TrendPoint) GetTimestampUnix() int64 {
@@ -571,7 +630,7 @@ type TrendResponse struct {

 func (x *TrendResponse) Reset() {
 	*x = TrendResponse{}
-	mi := &file_logtail_proto_msgTypes[6]
+	mi := &file_proto_logtail_proto_msgTypes[6]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -583,7 +642,7 @@ func (x *TrendResponse) String() string {
 func (*TrendResponse) ProtoMessage() {}

 func (x *TrendResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_logtail_proto_msgTypes[6]
+	mi := &file_proto_logtail_proto_msgTypes[6]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -596,7 +655,7 @@ func (x *TrendResponse) ProtoReflect() protoreflect.Message {

 // Deprecated: Use TrendResponse.ProtoReflect.Descriptor instead.
 func (*TrendResponse) Descriptor() ([]byte, []int) {
-	return file_logtail_proto_rawDescGZIP(), []int{6}
+	return file_proto_logtail_proto_rawDescGZIP(), []int{6}
 }

 func (x *TrendResponse) GetPoints() []*TrendPoint {
@@ -621,7 +680,7 @@ type SnapshotRequest struct {

 func (x *SnapshotRequest) Reset() {
 	*x = SnapshotRequest{}
-	mi := &file_logtail_proto_msgTypes[7]
+	mi := &file_proto_logtail_proto_msgTypes[7]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -633,7 +692,7 @@ func (x *SnapshotRequest) String() string {
 func (*SnapshotRequest) ProtoMessage() {}

 func (x *SnapshotRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_logtail_proto_msgTypes[7]
+	mi := &file_proto_logtail_proto_msgTypes[7]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -646,7 +705,7 @@ func (x *SnapshotRequest) ProtoReflect() protoreflect.Message {

 // Deprecated: Use SnapshotRequest.ProtoReflect.Descriptor instead.
 func (*SnapshotRequest) Descriptor() ([]byte, []int) {
-	return file_logtail_proto_rawDescGZIP(), []int{7}
+	return file_proto_logtail_proto_rawDescGZIP(), []int{7}
 }

 type Snapshot struct {
@@ -660,7 +719,7 @@ type Snapshot struct {

 func (x *Snapshot) Reset() {
 	*x = Snapshot{}
-	mi := &file_logtail_proto_msgTypes[8]
+	mi := &file_proto_logtail_proto_msgTypes[8]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -672,7 +731,7 @@ func (x *Snapshot) String() string {
 func (*Snapshot) ProtoMessage() {}

 func (x *Snapshot) ProtoReflect() protoreflect.Message {
-	mi := &file_logtail_proto_msgTypes[8]
+	mi := &file_proto_logtail_proto_msgTypes[8]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -685,7 +744,7 @@ func (x *Snapshot) ProtoReflect() protoreflect.Message {

 // Deprecated: Use Snapshot.ProtoReflect.Descriptor instead.
 func (*Snapshot) Descriptor() ([]byte, []int) {
-	return file_logtail_proto_rawDescGZIP(), []int{8}
+	return file_proto_logtail_proto_rawDescGZIP(), []int{8}
 }

 func (x *Snapshot) GetSource() string {
@@ -717,7 +776,7 @@ type ListTargetsRequest struct {

 func (x *ListTargetsRequest) Reset() {
 	*x = ListTargetsRequest{}
-	mi := &file_logtail_proto_msgTypes[9]
+	mi := &file_proto_logtail_proto_msgTypes[9]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -729,7 +788,7 @@ func (x *ListTargetsRequest) String() string {
 func (*ListTargetsRequest) ProtoMessage() {}

 func (x *ListTargetsRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_logtail_proto_msgTypes[9]
+	mi := &file_proto_logtail_proto_msgTypes[9]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -742,7 +801,7 @@ func (x *ListTargetsRequest) ProtoReflect() protoreflect.Message {

 // Deprecated: Use ListTargetsRequest.ProtoReflect.Descriptor instead.
 func (*ListTargetsRequest) Descriptor() ([]byte, []int) {
-	return file_logtail_proto_rawDescGZIP(), []int{9}
+	return file_proto_logtail_proto_rawDescGZIP(), []int{9}
 }

 type TargetInfo struct {
@@ -755,7 +814,7 @@ type TargetInfo struct {

 func (x *TargetInfo) Reset() {
 	*x = TargetInfo{}
-	mi := &file_logtail_proto_msgTypes[10]
+	mi := &file_proto_logtail_proto_msgTypes[10]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -767,7 +826,7 @@ func (x *TargetInfo) String() string {
 func (*TargetInfo) ProtoMessage() {}

 func (x *TargetInfo) ProtoReflect() protoreflect.Message {
-	mi := &file_logtail_proto_msgTypes[10]
+	mi := &file_proto_logtail_proto_msgTypes[10]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -780,7 +839,7 @@ func (x *TargetInfo) ProtoReflect() protoreflect.Message {

 // Deprecated: Use TargetInfo.ProtoReflect.Descriptor instead.
 func (*TargetInfo) Descriptor() ([]byte, []int) {
-	return file_logtail_proto_rawDescGZIP(), []int{10}
+	return file_proto_logtail_proto_rawDescGZIP(), []int{10}
 }

 func (x *TargetInfo) GetName() string {
@@ -806,7 +865,7 @@ type ListTargetsResponse struct {

 func (x *ListTargetsResponse) Reset() {
 	*x = ListTargetsResponse{}
-	mi := &file_logtail_proto_msgTypes[11]
+	mi := &file_proto_logtail_proto_msgTypes[11]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -818,7 +877,7 @@ func (x *ListTargetsResponse) String() string {
 func (*ListTargetsResponse) ProtoMessage() {}

 func (x *ListTargetsResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_logtail_proto_msgTypes[11]
+	mi := &file_proto_logtail_proto_msgTypes[11]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -831,7 +890,7 @@ func (x *ListTargetsResponse) ProtoReflect() protoreflect.Message {

 // Deprecated: Use ListTargetsResponse.ProtoReflect.Descriptor instead.
 func (*ListTargetsResponse) Descriptor() ([]byte, []int) {
-	return file_logtail_proto_rawDescGZIP(), []int{11}
+	return file_proto_logtail_proto_rawDescGZIP(), []int{11}
 }

 func (x *ListTargetsResponse) GetTargets() []*TargetInfo {
@@ -841,11 +900,11 @@ func (x *ListTargetsResponse) GetTargets() []*TargetInfo {
 	return nil
 }

-var File_logtail_proto protoreflect.FileDescriptor
+var File_proto_logtail_proto protoreflect.FileDescriptor

-const file_logtail_proto_rawDesc = "" +
+const file_proto_logtail_proto_rawDesc = "" +
 	"\n" +
-	"\rlogtail.proto\x12\alogtail\"\x8b\x03\n" +
+	"\x13proto/logtail.proto\x12\alogtail\"\xb1\x03\n" +
 	"\x06Filter\x12\x1d\n" +
 	"\awebsite\x18\x01 \x01(\tH\x00R\awebsite\x88\x01\x01\x12(\n" +
 	"\rclient_prefix\x18\x02 \x01(\tH\x01R\fclientPrefix\x88\x01\x01\x12-\n" +
@@ -853,7 +912,8 @@ const file_logtail_proto_rawDesc = "" +
 	"\rhttp_response\x18\x04 \x01(\x05H\x03R\fhttpResponse\x88\x01\x01\x12.\n" +
 	"\tstatus_op\x18\x05 \x01(\x0e2\x11.logtail.StatusOpR\bstatusOp\x12(\n" +
 	"\rwebsite_regex\x18\x06 \x01(\tH\x04R\fwebsiteRegex\x88\x01\x01\x12 \n" +
-	"\turi_regex\x18\a \x01(\tH\x05R\buriRegex\x88\x01\x01B\n" +
+	"\turi_regex\x18\a \x01(\tH\x05R\buriRegex\x88\x01\x01\x12$\n" +
+	"\x03tor\x18\b \x01(\x0e2\x12.logtail.TorFilterR\x03torB\n" +
 	"\n" +
 	"\b_websiteB\x10\n" +
 	"\x0e_client_prefixB\x13\n" +
@@ -894,7 +954,12 @@ const file_logtail_proto_rawDesc = "" +
 	"\x04name\x18\x01 \x01(\tR\x04name\x12\x12\n" +
 	"\x04addr\x18\x02 \x01(\tR\x04addr\"D\n" +
 	"\x13ListTargetsResponse\x12-\n" +
-	"\atargets\x18\x01 \x03(\v2\x13.logtail.TargetInfoR\atargets*:\n" +
+	"\atargets\x18\x01 \x03(\v2\x13.logtail.TargetInfoR\atargets*1\n" +
+	"\tTorFilter\x12\v\n" +
+	"\aTOR_ANY\x10\x00\x12\v\n" +
+	"\aTOR_YES\x10\x01\x12\n" +
+	"\n" +
+	"\x06TOR_NO\x10\x02*:\n" +
 	"\bStatusOp\x12\x06\n" +
 	"\x02EQ\x10\x00\x12\x06\n" +
 	"\x02NE\x10\x01\x12\x06\n" +
@@ -921,84 +986,86 @@ const file_logtail_proto_rawDesc = "" +
 	"\vListTargets\x12\x1b.logtail.ListTargetsRequest\x1a\x1c.logtail.ListTargetsResponseB0Z.git.ipng.ch/ipng/nginx-logtail/proto/logtailpbb\x06proto3"

 var (
-	file_logtail_proto_rawDescOnce sync.Once
-	file_logtail_proto_rawDescData []byte
+	file_proto_logtail_proto_rawDescOnce sync.Once
+	file_proto_logtail_proto_rawDescData []byte
 )

-func file_logtail_proto_rawDescGZIP() []byte {
-	file_logtail_proto_rawDescOnce.Do(func() {
-		file_logtail_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_logtail_proto_rawDesc), len(file_logtail_proto_rawDesc)))
+func file_proto_logtail_proto_rawDescGZIP() []byte {
+	file_proto_logtail_proto_rawDescOnce.Do(func() {
+		file_proto_logtail_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_proto_logtail_proto_rawDesc), len(file_proto_logtail_proto_rawDesc)))
 	})
-	return file_logtail_proto_rawDescData
+	return file_proto_logtail_proto_rawDescData
 }

-var file_logtail_proto_enumTypes = make([]protoimpl.EnumInfo, 3)
-var file_logtail_proto_msgTypes = make([]protoimpl.MessageInfo, 12)
-var file_logtail_proto_goTypes = []any{
-	(StatusOp)(0),               // 0: logtail.StatusOp
-	(GroupBy)(0),                // 1: logtail.GroupBy
-	(Window)(0),                 // 2: logtail.Window
-	(*Filter)(nil),              // 3: logtail.Filter
-	(*TopNRequest)(nil),         // 4: logtail.TopNRequest
-	(*TopNEntry)(nil),           // 5: logtail.TopNEntry
-	(*TopNResponse)(nil),        // 6: logtail.TopNResponse
-	(*TrendRequest)(nil),        // 7: logtail.TrendRequest
-	(*TrendPoint)(nil),          // 8: logtail.TrendPoint
-	(*TrendResponse)(nil),       // 9: logtail.TrendResponse
-	(*SnapshotRequest)(nil),     // 10: logtail.SnapshotRequest
-	(*Snapshot)(nil),            // 11: logtail.Snapshot
-	(*ListTargetsRequest)(nil),  // 12: logtail.ListTargetsRequest
-	(*TargetInfo)(nil),          // 13: logtail.TargetInfo
-	(*ListTargetsResponse)(nil), // 14: logtail.ListTargetsResponse
+var file_proto_logtail_proto_enumTypes = make([]protoimpl.EnumInfo, 4)
+var file_proto_logtail_proto_msgTypes = make([]protoimpl.MessageInfo, 12)
+var file_proto_logtail_proto_goTypes = []any{
+	(TorFilter)(0),              // 0: logtail.TorFilter
+	(StatusOp)(0),               // 1: logtail.StatusOp
+	(GroupBy)(0),                // 2: logtail.GroupBy
+	(Window)(0),                 // 3: logtail.Window
+	(*Filter)(nil),              // 4: logtail.Filter
+	(*TopNRequest)(nil),         // 5: logtail.TopNRequest
+	(*TopNEntry)(nil),           // 6: logtail.TopNEntry
+	(*TopNResponse)(nil),        // 7: logtail.TopNResponse
+	(*TrendRequest)(nil),        // 8: logtail.TrendRequest
+	(*TrendPoint)(nil),          // 9: logtail.TrendPoint
+	(*TrendResponse)(nil),       // 10: logtail.TrendResponse
+	(*SnapshotRequest)(nil),     // 11: logtail.SnapshotRequest
+	(*Snapshot)(nil),            // 12: logtail.Snapshot
+	(*ListTargetsRequest)(nil),  // 13: logtail.ListTargetsRequest
+	(*TargetInfo)(nil),          // 14: logtail.TargetInfo
+	(*ListTargetsResponse)(nil), // 15: logtail.ListTargetsResponse
 }
-var file_logtail_proto_depIdxs = []int32{
-	0,  // 0: logtail.Filter.status_op:type_name -> logtail.StatusOp
-	3,  // 1: logtail.TopNRequest.filter:type_name -> logtail.Filter
-	1,  // 2: logtail.TopNRequest.group_by:type_name -> logtail.GroupBy
-	2,  // 3: logtail.TopNRequest.window:type_name -> logtail.Window
-	5,  // 4: logtail.TopNResponse.entries:type_name -> logtail.TopNEntry
-	3,  // 5: logtail.TrendRequest.filter:type_name -> logtail.Filter
-	2,  // 6: logtail.TrendRequest.window:type_name -> logtail.Window
-	8,  // 7: logtail.TrendResponse.points:type_name -> logtail.TrendPoint
-	5,  // 8: logtail.Snapshot.entries:type_name -> logtail.TopNEntry
-	13, // 9: logtail.ListTargetsResponse.targets:type_name -> logtail.TargetInfo
-	4,  // 10: logtail.LogtailService.TopN:input_type -> logtail.TopNRequest
-	7,  // 11: logtail.LogtailService.Trend:input_type -> logtail.TrendRequest
-	10, // 12: logtail.LogtailService.StreamSnapshots:input_type -> logtail.SnapshotRequest
-	12, // 13: logtail.LogtailService.ListTargets:input_type -> logtail.ListTargetsRequest
-	6,  // 14: logtail.LogtailService.TopN:output_type -> logtail.TopNResponse
-	9,  // 15: logtail.LogtailService.Trend:output_type -> logtail.TrendResponse
-	11, // 16: logtail.LogtailService.StreamSnapshots:output_type -> logtail.Snapshot
-	14, // 17: logtail.LogtailService.ListTargets:output_type -> logtail.ListTargetsResponse
-	14, // [14:18] is the sub-list for method output_type
-	10, // [10:14] is the sub-list for method input_type
-	10, // [10:10] is the sub-list for extension type_name
-	10, // [10:10] is the sub-list for extension extendee
-	0,  // [0:10] is the sub-list for field type_name
+var file_proto_logtail_proto_depIdxs = []int32{
+	1,  // 0: logtail.Filter.status_op:type_name -> logtail.StatusOp
+	0,  // 1: logtail.Filter.tor:type_name -> logtail.TorFilter
+	4,  // 2: logtail.TopNRequest.filter:type_name -> logtail.Filter
+	2,  // 3: logtail.TopNRequest.group_by:type_name -> logtail.GroupBy
+	3,  // 4: logtail.TopNRequest.window:type_name -> logtail.Window
+	6,  // 5: logtail.TopNResponse.entries:type_name -> logtail.TopNEntry
+	4,  // 6: logtail.TrendRequest.filter:type_name -> logtail.Filter
+	3,  // 7: logtail.TrendRequest.window:type_name -> logtail.Window
+	9,  // 8: logtail.TrendResponse.points:type_name -> logtail.TrendPoint
+	6,  // 9: logtail.Snapshot.entries:type_name -> logtail.TopNEntry
+	14, // 10: logtail.ListTargetsResponse.targets:type_name -> logtail.TargetInfo
+	5,  // 11: logtail.LogtailService.TopN:input_type -> logtail.TopNRequest
+	8,  // 12: logtail.LogtailService.Trend:input_type -> logtail.TrendRequest
+	11, // 13: logtail.LogtailService.StreamSnapshots:input_type -> logtail.SnapshotRequest
+	13, // 14: logtail.LogtailService.ListTargets:input_type -> logtail.ListTargetsRequest
+	7,  // 15: logtail.LogtailService.TopN:output_type -> logtail.TopNResponse
+	10, // 16: logtail.LogtailService.Trend:output_type -> logtail.TrendResponse
+	12, // 17: logtail.LogtailService.StreamSnapshots:output_type -> logtail.Snapshot
+	15, // 18: logtail.LogtailService.ListTargets:output_type -> logtail.ListTargetsResponse
+	15, // [15:19] is the sub-list for method output_type
+	11, // [11:15] is the sub-list for method input_type
+	11, // [11:11] is the sub-list for extension type_name
+	11, // [11:11] is the sub-list for extension extendee
+	0,  // [0:11] is the sub-list for field type_name
 }

-func init() { file_logtail_proto_init() }
-func file_logtail_proto_init() {
-	if File_logtail_proto != nil {
+func init() { file_proto_logtail_proto_init() }
+func file_proto_logtail_proto_init() {
+	if File_proto_logtail_proto != nil {
 		return
 	}
-	file_logtail_proto_msgTypes[0].OneofWrappers = []any{}
+	file_proto_logtail_proto_msgTypes[0].OneofWrappers = []any{}
 	type x struct{}
 	out := protoimpl.TypeBuilder{
 		File: protoimpl.DescBuilder{
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
-			RawDescriptor: unsafe.Slice(unsafe.StringData(file_logtail_proto_rawDesc), len(file_logtail_proto_rawDesc)),
-			NumEnums:      3,
+			RawDescriptor: unsafe.Slice(unsafe.StringData(file_proto_logtail_proto_rawDesc), len(file_proto_logtail_proto_rawDesc)),
+			NumEnums:      4,
 			NumMessages:   12,
 			NumExtensions: 0,
 			NumServices:   1,
 		},
-		GoTypes:           file_logtail_proto_goTypes,
-		DependencyIndexes: file_logtail_proto_depIdxs,
-		EnumInfos:         file_logtail_proto_enumTypes,
-		MessageInfos:      file_logtail_proto_msgTypes,
+		GoTypes:           file_proto_logtail_proto_goTypes,
+		DependencyIndexes: file_proto_logtail_proto_depIdxs,
+		EnumInfos:         file_proto_logtail_proto_enumTypes,
+		MessageInfos:      file_proto_logtail_proto_msgTypes,
 	}.Build()
-	File_logtail_proto = out.File
-	file_logtail_proto_goTypes = nil
-	file_logtail_proto_depIdxs = nil
+	File_proto_logtail_proto = out.File
+	file_proto_logtail_proto_goTypes = nil
+	file_proto_logtail_proto_depIdxs = nil
 }
--- a/proto/logtailpb/logtail_grpc.pb.go
+++ b/proto/logtailpb/logtail_grpc.pb.go
@@ -2,7 +2,7 @@
 // versions:
 // - protoc-gen-go-grpc v1.6.1
 // - protoc             v3.21.12
-// source: logtail.proto
+// source: proto/logtail.proto

 package logtailpb

@@ -235,5 +235,5 @@ var LogtailService_ServiceDesc = grpc.ServiceDesc{
 			ServerStreams: true,
 		},
 	},
-	Metadata: "logtail.proto",
+	Metadata: "proto/logtail.proto",
 }
Author	SHA1	Message	Date
Pim van Pelt	0fb84813a5	Enable lame-duck on exit, so systemctl can 'restart' collectors	2026-03-23 22:18:05 +01:00
Pim van Pelt	cd7f15afaf	Add is_tor plumbing from collector->aggregator->frontend/cli	2026-03-23 22:17:39 +01:00
Pim van Pelt	b89caa594c	Auto-rediscover new glob patterns	2026-03-23 20:39:12 +01:00