Skip to content

feat(metrics): read the PostHog metrics3 schema #5

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
frankh merged 3 commits into from
Sep 15, 2026
Merged

feat(metrics): read the PostHog metrics3 schema #5

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
865f322
feat(metrics): read the PostHog metrics3 schema
frankh Sep 15, 2026
0598d0a
fix(metrics): preserve filtered label discovery
frankh Sep 15, 2026
e518d76
feat(metrics): serve filtered label discovery from the attribute rollup
frankh Sep 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 27 additions & 7 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -338,7 +338,7 @@ Metrics and logs choose their layouts independently.
| Data | Layout | Best for | Schema |
| --- | --- | --- | --- |
| Metrics | `current` (default) | New Snuffle deployments optimized around Prometheus series, samples, labels, histograms, exemplars, and metadata | [`scripts/create_metrics_schema.sql`](scripts/create_metrics_schema.sql) |
| Metrics | `posthog` | Existing PostHog-style `metrics2`, `metric_series2`, and `metric_attributes2` tables | [`scripts/create_metrics_posthog_schema.sql`](scripts/create_metrics_posthog_schema.sql) |
| Metrics | `posthog` | Existing PostHog-style `metrics2`, `metric_series3`, `metric_attributes3`, and `metric_names3` tables | [`scripts/create_metrics_posthog_schema.sql`](scripts/create_metrics_posthog_schema.sql) |
| Logs | `snuffle` (default with `current` metrics) | New deployments with a narrow log table, stream dictionary, label index, and minute rollups | [`scripts/create_logs_snuffle_schema.sql`](scripts/create_logs_snuffle_schema.sql) |
| Logs | `posthog` (default with `posthog` metrics) | Existing PostHog-style `logs34` and `log_attributes3` tables | [`scripts/create_logs_posthog_schema.sql`](scripts/create_logs_posthog_schema.sql) |

Expand Down Expand Up @@ -372,11 +372,29 @@ the hot log table narrow while retaining selector and aggregation support.
### PostHog-compatible data

In PostHog metrics mode, series identity is the `series_fingerprint` shared by
`metric_series2` and `metrics2`. Snuffle selects series from `metric_series2`,
`metric_series3` and `metrics2`. Snuffle selects series from `metric_series3`,
builds Prometheus labels from `metric_name`, `service_name`,
`resource_attributes`, and `attributes`, and reads samples from `metrics2` by
fingerprint. Remote write inserts into `metrics2_input`; its materialized views
fan each row out to the samples, series, and attribute tables.
fingerprint. `metric_series3` keeps one row per series and expiry day, so
series reads collapse duplicates by fingerprint. Label discovery reads the
hourly rollups instead of the series table: `metric_names3` lists metric
names, and `metric_attributes3` lists attribute keys and values, filtered by
exact `__name__` and `service_name` matchers. Other matchers fall back to the
series table. Remote write inserts into `metrics2_input`; its materialized
views fan each row out to the samples, series, attribute, and name tables.

The rollups set two limits on the Prometheus label surface:

- Attribute keys and values must be shorter than 256 characters. The
attribute rollup drops longer pairs, so label discovery does not list them.
- When a key exists in both `resource_attributes` and `attributes`, the
Prometheus label carries the resource attribute value. Label value
discovery for that key can also list the metric attribute value.

Set `CH_SERIES_TABLE=metric_series2`, `CH_ATTRIBUTE_TABLE=metric_attributes2`,
`CH_ATTRIBUTE_TABLE_HAS_METRIC_NAME=false`, and an empty
`CH_METRIC_NAMES_TABLE` to read the previous PostHog tables, for example
before their backfill into the `3` tables is complete.

In PostHog logs mode, Loki stream labels and structured metadata map onto the
OpenTelemetry-shaped `logs34` columns. Service, severity, trace, span, resource,
Expand Down Expand Up @@ -440,11 +458,13 @@ Snuffle is configured with environment variables.
| Variable | Default | Purpose |
| --- | --- | --- |
| `CH_SCHEMA_LAYOUT` | `current` | Metrics layout: `current` or `posthog`; `SNUFFLE_SCHEMA_LAYOUT` is accepted as a legacy fallback |
| `CH_SERIES_TABLE` | `metrics_series` / `metric_series2` | Series table |
| `CH_SERIES_TABLE` | `metrics_series` / `metric_series3` | Series table |
| `CH_SAMPLES_TABLE` | `metrics_samples` / `metrics2` | Float sample table |
| `CH_LABEL_INDEX_TABLE` | `metrics_label_index` / empty | Metrics label index |
| `CH_ATTRIBUTE_TABLE` | `metric_attributes` / `metric_attributes2` | PostHog attribute discovery table |
| `CH_METRICS_INPUT_TABLE` | empty / `metrics2_input` | PostHog remote write target; its materialized views feed the samples, series, and attribute tables |
| `CH_ATTRIBUTE_TABLE` | `metric_attributes` / `metric_attributes3` | PostHog attribute discovery table |
| `CH_METRIC_NAMES_TABLE` | empty / `metric_names3` | PostHog metric name discovery table; empty reads metric names from the series table |
| `CH_ATTRIBUTE_TABLE_HAS_METRIC_NAME` | `false` / `true` | Whether the PostHog attribute table has a `metric_name` column; set `false` with `metric_attributes2` |
| `CH_METRICS_INPUT_TABLE` | empty / `metrics2_input` | PostHog remote write target; its materialized views feed the samples, series, attribute, and name tables |
| `CH_LABEL_POSTINGS_TABLE` | empty | Optional optimized metrics postings table |
| `CH_ACTIVITY_TABLE` | empty | Optional series-activity table |
| `CH_METRICS_TABLE` | `metrics_metadata` / empty | Metric metadata table |
Expand Down
2 changes: 1 addition & 1 deletion internal/snuffle/api.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,7 @@ func Run(cfg Config) error {
ReadHeaderTimeout: 5 * time.Second,
TLSConfig: tlsConfig,
}
slog.Info("starting PromQL ClickHouse sidecar", "addr", addr, "tls", tlsConfig != nil, "schema_layout", cfg.SchemaLayout, "series_table", cfg.SeriesTable, "samples_table", cfg.SamplesTable, "label_index_table", cfg.LabelIndexTable, "lookback_delta", cfg.LookbackDelta.String(), "max_samples", cfg.MaxSamples)
slog.Info("starting PromQL ClickHouse sidecar", "addr", addr, "tls", tlsConfig != nil, "schema_layout", cfg.SchemaLayout, "series_table", cfg.SeriesTable, "samples_table", cfg.SamplesTable, "label_index_table", cfg.LabelIndexTable, "attribute_table", cfg.AttributeTable, "metric_names_table", cfg.MetricNamesTable, "lookback_delta", cfg.LookbackDelta.String(), "max_samples", cfg.MaxSamples)
if tlsConfig != nil {
return srv.ListenAndServeTLS("", "")
}
Expand Down
192 changes: 102 additions & 90 deletions internal/snuffle/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,45 +18,49 @@ type Config struct {
SamplesTable string
LabelIndexTable string
AttributeTable string
LabelPostingsTable string
ActivityTable string
MetricsTable string
MetricsInputTable string
LogSchemaLayout string
LogsTable string
LogStreamsTable string
LogStreamLabelsTable string
LogAttributesTable string
LogStreamStatsTable string
HistogramsTable string
ExemplarsTable string
HTTPHost string
HTTPPort string
TLSEnabled bool
TLSCertFile string
TLSKeyFile string
CHTimeout time.Duration
QueryTimeout time.Duration
LookbackDelta time.Duration
MaxSamples int
MaxSeries int
IDChunkSize int
AggregateThreads int
RemoteWriteInterval time.Duration
SampleAttributes bool
TeamID uint64
DefaultTeamID uint64
TeamHeader string
TeamQueryParam string
Pprof bool
SelfScrapeEnabled bool
SelfScrapeInterval time.Duration
SelfScrapeTeamID uint64
SelfScrapeJob string
SelfScrapeInstance string
LogRetention time.Duration
LogQueryMaxRows int
MetricsRetention time.Duration
MetricNamesTable string
// AttributeTableHasMetricName allows filtered label discovery to read the
// attribute table. The legacy metric_attributes2 table has no metric_name.
AttributeTableHasMetricName bool
LabelPostingsTable string
ActivityTable string
MetricsTable string
MetricsInputTable string
LogSchemaLayout string
LogsTable string
LogStreamsTable string
LogStreamLabelsTable string
LogAttributesTable string
LogStreamStatsTable string
HistogramsTable string
ExemplarsTable string
HTTPHost string
HTTPPort string
TLSEnabled bool
TLSCertFile string
TLSKeyFile string
CHTimeout time.Duration
QueryTimeout time.Duration
LookbackDelta time.Duration
MaxSamples int
MaxSeries int
IDChunkSize int
AggregateThreads int
RemoteWriteInterval time.Duration
SampleAttributes bool
TeamID uint64
DefaultTeamID uint64
TeamHeader string
TeamQueryParam string
Pprof bool
SelfScrapeEnabled bool
SelfScrapeInterval time.Duration
SelfScrapeTeamID uint64
SelfScrapeJob string
SelfScrapeInstance string
LogRetention time.Duration
LogQueryMaxRows int
MetricsRetention time.Duration
}

func ConfigFromEnv() Config {
Expand All @@ -78,6 +82,7 @@ func ConfigFromEnv() Config {
exemplarsTableDefault := "metrics_exemplars"
metadataTableDefault := "metrics_metadata"
attributeTableDefault := "metric_attributes"
metricNamesTableDefault := ""
metricsInputTableDefault := ""
logsTableDefault := "logs"
logStreamsTableDefault := "log_streams"
Expand All @@ -86,10 +91,11 @@ func ConfigFromEnv() Config {
logStreamStatsTableDefault := "log_stream_stats"
aggregateThreadsDefault := 1
if schemaLayout == schemaLayoutPostHog {
seriesTableDefault = "metric_series2"
seriesTableDefault = "metric_series3"
samplesTableDefault = "metrics2"
labelIndexTableDefault = ""
attributeTableDefault = "metric_attributes2"
attributeTableDefault = "metric_attributes3"
metricNamesTableDefault = "metric_names3"
metricsInputTableDefault = "metrics2_input"
histogramsTableDefault = ""
exemplarsTableDefault = ""
Expand All @@ -102,56 +108,62 @@ func ConfigFromEnv() Config {
logAttributesTableDefault = "log_attributes3"
logStreamStatsTableDefault = ""
}
metricNamesTable, metricNamesTableSet := os.LookupEnv("CH_METRIC_NAMES_TABLE")
if !metricNamesTableSet {
metricNamesTable = metricNamesTableDefault
}
return Config{
CHAddr: getenv("CH_ADDR", "localhost:9000"),
CHUser: getenv("CH_USER", "default"),
CHPassword: os.Getenv("CH_PASSWORD"),
AllowUnauthenticated: envBool("SNUFFLE_ALLOW_UNAUTHENTICATED", false),
CHDatabase: getenv("CH_DATABASE", "default"),
SchemaLayout: string(schemaLayout),
SeriesTable: getenv("CH_SERIES_TABLE", getenv("CH_TAGS_TABLE", seriesTableDefault)),
SamplesTable: getenv("CH_SAMPLES_TABLE", getenv("CH_DATA_TABLE", samplesTableDefault)),
LabelIndexTable: getenv("CH_LABEL_INDEX_TABLE", labelIndexTableDefault),
AttributeTable: getenv("CH_ATTRIBUTE_TABLE", attributeTableDefault),
LabelPostingsTable: getenv("CH_LABEL_POSTINGS_TABLE", ""),
ActivityTable: getenv("CH_ACTIVITY_TABLE", ""),
MetricsTable: getenv("CH_METRICS_TABLE", metadataTableDefault),
MetricsInputTable: getenv("CH_METRICS_INPUT_TABLE", metricsInputTableDefault),
LogSchemaLayout: string(logSchemaLayout),
LogsTable: getenv("CH_LOGS_TABLE", logsTableDefault),
LogStreamsTable: getenv("CH_LOG_STREAMS_TABLE", logStreamsTableDefault),
LogStreamLabelsTable: getenv("CH_LOG_STREAM_LABELS_TABLE", logStreamLabelsTableDefault),
LogAttributesTable: getenv("CH_LOG_ATTRIBUTES_TABLE", getenv("CH_LOG_ATTRIBUTE_TABLE", logAttributesTableDefault)),
LogStreamStatsTable: getenv("CH_LOG_STREAM_STATS_TABLE", logStreamStatsTableDefault),
HistogramsTable: getenv("CH_HISTOGRAMS_TABLE", histogramsTableDefault),
ExemplarsTable: getenv("CH_EXEMPLARS_TABLE", exemplarsTableDefault),
HTTPHost: getenv("SIDECAR_HOST", "0.0.0.0"),
HTTPPort: httpPort,
TLSEnabled: envBool("SNUFFLE_TLS_ENABLED", tlsCertFile != "" || tlsKeyFile != ""),
TLSCertFile: tlsCertFile,
TLSKeyFile: tlsKeyFile,
CHTimeout: envDurationSeconds("CH_TIMEOUT_SECONDS", 30*time.Second),
QueryTimeout: envDurationSeconds("PROMQL_QUERY_TIMEOUT_SECONDS", 30*time.Second),
LookbackDelta: envDuration("PROMQL_LOOKBACK_DELTA", 5*time.Minute),
MaxSamples: envInt("PROMQL_MAX_SAMPLES", 50_000_000, 1),
MaxSeries: envInt("CH_MAX_SERIES", 1_000_000, 1),
IDChunkSize: envInt("CH_ID_CHUNK_SIZE", 20000, 1),
AggregateThreads: envInt("CH_AGGREGATE_MAX_THREADS", aggregateThreadsDefault, 0),
RemoteWriteInterval: envDurationAllowZero("REMOTE_WRITE_SAMPLE_INTERVAL", 15*time.Second),
SampleAttributes: envBool("SNUFFLE_SAMPLE_ATTRIBUTES", sampleAttributesDefault),
TeamID: defaultTeamID,
DefaultTeamID: defaultTeamID,
TeamHeader: getenv("SNUFFLE_TEAM_HEADER", "X-Team-ID"),
TeamQueryParam: getenv("SNUFFLE_TEAM_QUERY_PARAM", "team_id"),
Pprof: envBool("SNUFFLE_PPROF", false),
SelfScrapeEnabled: envBool("SNUFFLE_SELF_SCRAPE_ENABLED", true),
SelfScrapeInterval: envDurationAllowZero("SNUFFLE_SELF_SCRAPE_INTERVAL", 15*time.Second),
SelfScrapeTeamID: envUint64("SNUFFLE_SELF_SCRAPE_TEAM_ID", defaultTeamID),
SelfScrapeJob: getenv("SNUFFLE_SELF_SCRAPE_JOB", "snuffle"),
SelfScrapeInstance: getenv("SNUFFLE_SELF_SCRAPE_INSTANCE", defaultSelfScrapeInstance(httpPort)),
LogRetention: envDuration("SNUFFLE_LOG_RETENTION", 30*24*time.Hour),
LogQueryMaxRows: envInt("SNUFFLE_LOG_QUERY_MAX_ROWS", 100000, 1),
MetricsRetention: envDuration("SNUFFLE_METRICS_RETENTION", 90*24*time.Hour),
CHAddr: getenv("CH_ADDR", "localhost:9000"),
CHUser: getenv("CH_USER", "default"),
CHPassword: os.Getenv("CH_PASSWORD"),
AllowUnauthenticated: envBool("SNUFFLE_ALLOW_UNAUTHENTICATED", false),
CHDatabase: getenv("CH_DATABASE", "default"),
SchemaLayout: string(schemaLayout),
SeriesTable: getenv("CH_SERIES_TABLE", getenv("CH_TAGS_TABLE", seriesTableDefault)),
SamplesTable: getenv("CH_SAMPLES_TABLE", getenv("CH_DATA_TABLE", samplesTableDefault)),
LabelIndexTable: getenv("CH_LABEL_INDEX_TABLE", labelIndexTableDefault),
AttributeTable: getenv("CH_ATTRIBUTE_TABLE", attributeTableDefault),
MetricNamesTable: metricNamesTable,
AttributeTableHasMetricName: envBool("CH_ATTRIBUTE_TABLE_HAS_METRIC_NAME", schemaLayout == schemaLayoutPostHog),
LabelPostingsTable: getenv("CH_LABEL_POSTINGS_TABLE", ""),
ActivityTable: getenv("CH_ACTIVITY_TABLE", ""),
MetricsTable: getenv("CH_METRICS_TABLE", metadataTableDefault),
MetricsInputTable: getenv("CH_METRICS_INPUT_TABLE", metricsInputTableDefault),
LogSchemaLayout: string(logSchemaLayout),
LogsTable: getenv("CH_LOGS_TABLE", logsTableDefault),
LogStreamsTable: getenv("CH_LOG_STREAMS_TABLE", logStreamsTableDefault),
LogStreamLabelsTable: getenv("CH_LOG_STREAM_LABELS_TABLE", logStreamLabelsTableDefault),
LogAttributesTable: getenv("CH_LOG_ATTRIBUTES_TABLE", getenv("CH_LOG_ATTRIBUTE_TABLE", logAttributesTableDefault)),
LogStreamStatsTable: getenv("CH_LOG_STREAM_STATS_TABLE", logStreamStatsTableDefault),
HistogramsTable: getenv("CH_HISTOGRAMS_TABLE", histogramsTableDefault),
ExemplarsTable: getenv("CH_EXEMPLARS_TABLE", exemplarsTableDefault),
HTTPHost: getenv("SIDECAR_HOST", "0.0.0.0"),
HTTPPort: httpPort,
TLSEnabled: envBool("SNUFFLE_TLS_ENABLED", tlsCertFile != "" || tlsKeyFile != ""),
TLSCertFile: tlsCertFile,
TLSKeyFile: tlsKeyFile,
CHTimeout: envDurationSeconds("CH_TIMEOUT_SECONDS", 30*time.Second),
QueryTimeout: envDurationSeconds("PROMQL_QUERY_TIMEOUT_SECONDS", 30*time.Second),
LookbackDelta: envDuration("PROMQL_LOOKBACK_DELTA", 5*time.Minute),
MaxSamples: envInt("PROMQL_MAX_SAMPLES", 50_000_000, 1),
MaxSeries: envInt("CH_MAX_SERIES", 1_000_000, 1),
IDChunkSize: envInt("CH_ID_CHUNK_SIZE", 20000, 1),
AggregateThreads: envInt("CH_AGGREGATE_MAX_THREADS", aggregateThreadsDefault, 0),
RemoteWriteInterval: envDurationAllowZero("REMOTE_WRITE_SAMPLE_INTERVAL", 15*time.Second),
SampleAttributes: envBool("SNUFFLE_SAMPLE_ATTRIBUTES", sampleAttributesDefault),
TeamID: defaultTeamID,
DefaultTeamID: defaultTeamID,
TeamHeader: getenv("SNUFFLE_TEAM_HEADER", "X-Team-ID"),
TeamQueryParam: getenv("SNUFFLE_TEAM_QUERY_PARAM", "team_id"),
Pprof: envBool("SNUFFLE_PPROF", false),
SelfScrapeEnabled: envBool("SNUFFLE_SELF_SCRAPE_ENABLED", true),
SelfScrapeInterval: envDurationAllowZero("SNUFFLE_SELF_SCRAPE_INTERVAL", 15*time.Second),
SelfScrapeTeamID: envUint64("SNUFFLE_SELF_SCRAPE_TEAM_ID", defaultTeamID),
SelfScrapeJob: getenv("SNUFFLE_SELF_SCRAPE_JOB", "snuffle"),
SelfScrapeInstance: getenv("SNUFFLE_SELF_SCRAPE_INSTANCE", defaultSelfScrapeInstance(httpPort)),
LogRetention: envDuration("SNUFFLE_LOG_RETENTION", 30*24*time.Hour),
LogQueryMaxRows: envInt("SNUFFLE_LOG_QUERY_MAX_ROWS", 100000, 1),
MetricsRetention: envDuration("SNUFFLE_METRICS_RETENTION", 90*24*time.Hour),
}
}

Expand Down
21 changes: 19 additions & 2 deletions internal/snuffle/config_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,8 +84,8 @@ func TestConfigFromEnvSchemaLayout(t *testing.T) {
if !cfg.SampleAttributes {
t.Fatalf("SampleAttributes = false, want true for posthog layout")
}
if cfg.SamplesTable != "metrics2" || cfg.SeriesTable != "metric_series2" || cfg.LabelIndexTable != "" || cfg.AttributeTable != "metric_attributes2" || cfg.MetricsInputTable != "metrics2_input" {
t.Fatalf("posthog tables = samples %q series %q label_index %q attributes %q input %q", cfg.SamplesTable, cfg.SeriesTable, cfg.LabelIndexTable, cfg.AttributeTable, cfg.MetricsInputTable)
if cfg.SamplesTable != "metrics2" || cfg.SeriesTable != "metric_series3" || cfg.LabelIndexTable != "" || cfg.AttributeTable != "metric_attributes3" || cfg.MetricNamesTable != "metric_names3" || cfg.MetricsInputTable != "metrics2_input" {
t.Fatalf("posthog tables = samples %q series %q label_index %q attributes %q names %q input %q", cfg.SamplesTable, cfg.SeriesTable, cfg.LabelIndexTable, cfg.AttributeTable, cfg.MetricNamesTable, cfg.MetricsInputTable)
}
if cfg.MetricsRetention != 90*24*time.Hour {
t.Fatalf("MetricsRetention = %s, want 90 days", cfg.MetricsRetention)
Expand All @@ -104,6 +104,23 @@ func TestConfigFromEnvSchemaLayout(t *testing.T) {
}
}

func TestConfigFromEnvMetricNamesTableOverride(t *testing.T) {
t.Setenv("CH_SCHEMA_LAYOUT", "posthog")
if !ConfigFromEnv().AttributeTableHasMetricName {
t.Fatalf("AttributeTableHasMetricName = false, want true for posthog layout")
}
t.Setenv("CH_ATTRIBUTE_TABLE_HAS_METRIC_NAME", "false")
if ConfigFromEnv().AttributeTableHasMetricName {
t.Fatalf("AttributeTableHasMetricName = true, want false when disabled")
}
for _, table := range []string{"", "custom_metric_names"} {
t.Setenv("CH_METRIC_NAMES_TABLE", table)
if got := ConfigFromEnv().MetricNamesTable; got != table {
t.Fatalf("MetricNamesTable = %q, want %q", got, table)
}
}
}

func TestConfigFromEnvSelfScrapeSettings(t *testing.T) {
t.Setenv("SNUFFLE_DEFAULT_TEAM_ID", "42")
t.Setenv("SNUFFLE_SELF_SCRAPE_INTERVAL", "30s")
Expand Down
Loading
Loading