feat: use clickhouse for general metrics

go/go.mod

@@ -0,0 +1,3 @@
+module github.com/unkeyed/unkey/go
+
+go 1.23.4

go/pkg/clickhouse/client.go

@@ -0,0 +1,105 @@
+package clickhouse
+
+import (
+	"context"
+	"time"
+
+	ch "github.com/ClickHouse/clickhouse-go/v2"
+	"github.com/Southclaws/fault"
+	"github.com/Southclaws/fault/fmsg"
+	"github.com/unkeyed/unkey/apps/agent/pkg/batch"
+	"github.com/unkeyed/unkey/apps/agent/pkg/clickhouse/schema"
+	"github.com/unkeyed/unkey/apps/agent/pkg/logging"
+	"github.com/unkeyed/unkey/apps/agent/pkg/util"
+)
+
+type Clickhouse struct {
+	conn   ch.Conn
+	logger logging.Logger
+
+	requests         *batch.BatchProcessor[schema.ApiRequestV1]
+	keyVerifications *batch.BatchProcessor[schema.KeyVerificationRequestV1]
+}
+
+type Config struct {
+	URL    string
+	Logger logging.Logger
+}
+
+func New(config Config) (*Clickhouse, error) {
+
+	opts, err := ch.ParseDSN(config.URL)
+	if err != nil {
+		return nil, fault.Wrap(err, fmsg.With("parsing clickhouse DSN failed"))
+	}
+
+	// opts.TLS = &tls.Config{}
+	opts.Debug = true
+	opts.Debugf = func(format string, v ...any) {
+		config.Logger.Debug().Msgf(format, v...)
+	}
+	config.Logger.Info().Interface("opts", opts.Addr).Msg("connecting to clickhouse")
+	conn, err := ch.Open(opts)
+	if err != nil {
+		return nil, fault.Wrap(err, fmsg.With("opening clickhouse failed"))
+	}
+
+	err = util.Retry(func() error {
+		return conn.Ping(context.Background())
+	}, 10, func(n int) time.Duration {
+		return time.Duration(n) * time.Second
+	})
+	if err != nil {
+		return nil, fault.Wrap(err, fmsg.With("pinging clickhouse failed"))
+	}
+	c := &Clickhouse{
+		conn:   conn,
+		logger: config.Logger,
+
+		requests: batch.New[schema.ApiRequestV1](batch.Config[schema.ApiRequestV1]{
+			BatchSize:     1000,
+			BufferSize:    100000,
+			FlushInterval: time.Second,
+			Consumers:     4,
+			Flush: func(ctx context.Context, rows []schema.ApiRequestV1) {
+				table := "raw_api_requests_v1"
+				err := flush(ctx, conn, table, rows)
+				if err != nil {
+					config.Logger.Error().Err(err).Str("table", table).Msg("failed to flush batch")
+				}
+			},
+		}),
+		keyVerifications: batch.New[schema.KeyVerificationRequestV1](batch.Config[schema.KeyVerificationRequestV1]{
+			BatchSize:     1000,
+			BufferSize:    100000,
+			FlushInterval: time.Second,
+			Consumers:     4,
+			Flush: func(ctx context.Context, rows []schema.KeyVerificationRequestV1) {
+				table := "raw_key_verifications_v1"
+				err := flush(ctx, conn, table, rows)
+				if err != nil {
+					config.Logger.Error().Err(err).Str("table", table).Msg("failed to flush batch")
+				}
+			},
+		}),
+	}
+
+	// err = c.conn.Ping(context.Background())
+	// if err != nil {
+	// 	return nil, fault.Wrap(err, fmsg.With("pinging clickhouse failed"))
+	// }
+	return c, nil
+}
+
+func (c *Clickhouse) Shutdown(ctx context.Context) error {
+	c.requests.Close()
+	return c.conn.Close()
+}
+
+func (c *Clickhouse) BufferApiRequest(req schema.ApiRequestV1) {
+	c.requests.Buffer(req)
+}
+
+func (c *Clickhouse) BufferKeyVerification(req schema.KeyVerificationRequestV1) {
+	c.keyVerifications.Buffer(req)
+}

go/pkg/clickhouse/flush.go

@@ -0,0 +1,28 @@
+package clickhouse
+
+import (
+	"context"
+	"fmt"
+
+	ch "github.com/ClickHouse/clickhouse-go/v2"
+	"github.com/Southclaws/fault"
+	"github.com/Southclaws/fault/fmsg"
+)
+
+func flush[T any](ctx context.Context, conn ch.Conn, table string, rows []T) error {
+	batch, err := conn.PrepareBatch(ctx, fmt.Sprintf("INSERT INTO %s", table))
+	if err != nil {
+		return fault.Wrap(err, fmsg.With("preparing batch failed"))
+	}
+	for _, row := range rows {
+		err = batch.AppendStruct(&row)
+		if err != nil {
+			return fault.Wrap(err, fmsg.With("appending struct to batch failed"))
+		}
+	}
+	err = batch.Send()
+	if err != nil {
+		return fault.Wrap(err, fmsg.With("committing batch failed"))
+	}
+	return nil
+}

go/pkg/clickhouse/interface.go

@@ -0,0 +1,10 @@
+package clickhouse
+
+import (
+	"github.com/unkeyed/unkey/apps/agent/pkg/clickhouse/schema"
+)
+
+type Bufferer interface {
+	BufferApiRequest(schema.ApiRequestV1)
+	BufferKeyVerification(schema.KeyVerificationRequestV1)
+}

go/pkg/clickhouse/noop.go

@@ -0,0 +1,20 @@
+package clickhouse
+
+import (
+	"github.com/unkeyed/unkey/apps/agent/pkg/clickhouse/schema"
+)
+
+type noop struct{}
+
+var _ Bufferer = &noop{}
+
+func (n *noop) BufferApiRequest(schema.ApiRequestV1) {
+
+}
+func (n *noop) BufferKeyVerification(schema.KeyVerificationRequestV1) {
+
+}
+
+func NewNoop() *noop {
+	return &noop{}
+}

go/pkg/clickhouse/schema/requests.go

@@ -0,0 +1,68 @@
+package schema
+
+type Sample struct {
+	// Temporality of the metric. This is used to identify the type of the metric. It can be one of
+	// the following values:
+	// - Unspecified: This is used for gauge metrics.
+	// - Cumulative: This is used for monotonic counters.
+	// - Delta: This is used for counters that reset after flushing.
+	//
+	// LowCardinality(String) DEFAULT 'Unspecified',
+	Temporality string `ch:"temporality"`
+
+	// Name of the metric.
+	//
+	// LowCardinality(String),
+	Name string `ch:"name"`
+
+	// Fingerprint of the metric. This is used to identify the metric uniquely. Currently, we are
+	// using the hash of the labels to generate the fingerprint.
+	//
+	// UInt64 CODEC(Delta(8), ZSTD(1)),
+	Fingerprint uint64 `ch:"fingerprint"`
+
+	// Timestamp in milliseconds when the metric was observed.
+	//
+	// Int64 CODEC(DoubleDelta, ZSTD(1)),
+	UnixMilli int64 `ch:"unix_milli"`
+
+	// Value of the metric.
+	//
+	// Float64 CODEC(Gorilla, ZSTD(1))
+	Value float64 `ch:"value"`
+}
+
+type Metric struct {
+	// Name of the metric.
+	//
+	// LowCardinality(String)
+	MetricName string
+
+	// Description of the metric.
+	//
+	// LowCardinality(String)
+	Description string
+
+	// Unit of the metric.
+	//
+	// For example: "s" for seconds
+	//
+	// LowCardinality(String)
+	Unit string
+
+	// Type: Type of the metric. One of the following values:
+	// - Sum: This is used for monotonic counters.
+	// - Gauge: This is used for gauge metrics.
+	// - Histogram: This is used for histogram metrics.
+	// - ExponentialHistogram: This is used for exponential histogram metrics.
+	Type string
+
+	// Fingerprint of the metric. This is used to identify the metric uniquely.
+	// Currently, we are using the hash of the labels to generate the fingerprint.
+	Fingerprint uint64
+
+	// Labels of the metric; Stored as a JSON string.
+	// The JSON string is sorted lexicographically to merge identical sets of labels using
+	// ClickHouse's ReplacingMergeTree.
+	Labels string
+}

package.json

@@ -31,6 +31,7 @@
     "knip": "^5.11.0",
     "lint-staged": "^15.2.2",
     "ts-node": "latest",
+    "tsc-shame": "^0.2.4",
     "turbo": "^2.1.3",
     "typescript": "^5.5.3",
     "vitest": "^1.6.0"