Documentation
¶
Overview ¶
Package analytics provides real-time search analytics collection and aggregation. Events are published to Kafka by the Collector and consumed by the Aggregator which maintains in-memory counters, latency histograms, and top-query rankings.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func HandleEvent ¶
func HandleEvent(agg *Aggregator) kafka.MessageHandler
HandleEvent returns a Kafka message handler that decodes SearchEvent or IndexEvent JSON payloads and records them in the aggregator.
Types ¶
type AggregatedStats ¶
type AggregatedStats struct {
TotalSearches int64 `json:"total_searches"`
TotalDocIndexed int64 `json:"total_docs_indexed"`
CacheHits int64 `json:"cache_hits"`
CacheMisses int64 `json:"cache_misses"`
ZeroResultCount int64 `json:"zero_result_count"`
AvgLatencyMs float64 `json:"avg_latency_ms"`
P50LatencyMs int64 `json:"p50_latency_ms"`
P95LatencyMs int64 `json:"p95_latency_ms"`
P99LatencyMs int64 `json:"p99_latency_ms"`
TopQueries []QueryCount `json:"top_queries"`
ZeroResultQueries []QueryCount `json:"zero_result_queries"`
QueriesPerMinute float64 `json:"queries_per_minute"`
}
AggregatedStats is the snapshot returned by Aggregator.Stats(). It contains search counts, document-indexing counts, cache hit/miss counters, latency percentiles, throughput, and the most-frequent queries.
type Aggregator ¶
type Aggregator struct {
// contains filtered or unexported fields
}
Aggregator consumes analytics events from Kafka and maintains an in-memory aggregate. It is safe for concurrent reads (Stats) and writes (record*).
func NewAggregator ¶
func NewAggregator(consumer *kafka.Consumer) *Aggregator
NewAggregator creates an Aggregator backed by the given Kafka consumer.
func (*Aggregator) Start ¶
func (a *Aggregator) Start(ctx context.Context) error
Start begins consuming events from Kafka. It blocks until ctx is cancelled.
func (*Aggregator) Stats ¶
func (a *Aggregator) Stats() AggregatedStats
Stats returns a point-in-time snapshot of all aggregated analytics data.
type Collector ¶
type Collector struct {
// contains filtered or unexported fields
}
Collector buffers analytics events in-memory and publishes them to Kafka asynchronously. If the internal channel fills up, events are dropped with a warning log rather than blocking the caller.
func NewCollector ¶
NewCollector creates a Collector with the given Kafka producer and channel buffer size. If bufferSize <= 0 it defaults to 10 000.
func (*Collector) Close ¶
func (c *Collector) Close()
Close shuts down the collector by closing the event channel and waiting for the background goroutine to finish draining.
type Handler ¶
type Handler struct {
// contains filtered or unexported fields
}
Handler exposes an HTTP endpoint for reading aggregated analytics.
func NewHandler ¶
func NewHandler(aggregator *Aggregator) *Handler
NewHandler creates a Handler backed by the given Aggregator.
type IndexEvent ¶
type IndexEvent struct {
Type EventType `json:"type"`
DocumentID string `json:"document_id"`
ShardID int `json:"shard_id"`
TokenCount int `json:"token_count"`
SizeBytes int `json:"size_bytes"`
LatencyMs int64 `json:"latency_ms"`
Timestamp time.Time `json:"timestamp"`
}
IndexEvent is emitted after a document is indexed into a shard.
type QueryCount ¶
QueryCount pairs a query string with how many times it has been executed.
type SearchEvent ¶
type SearchEvent struct {
Type EventType `json:"type"`
Query string `json:"query"`
Terms []string `json:"terms"`
TotalHits int `json:"total_hits"`
Returned int `json:"returned"`
LatencyMs int64 `json:"latency_ms"`
CacheHit bool `json:"cache_hit"`
ShardCount int `json:"shard_count"`
Timestamp time.Time `json:"timestamp"`
RequestID string `json:"request_id"`
}
SearchEvent is emitted by the search handler after each query and records the query text, result count, latency, cache status, and shard count.
Source Files
Directories
¶
| Path | Synopsis |
|---|---|
|
Package aggregator provides persistent storage and periodic snapshotting of aggregated analytics stats to PostgreSQL.
|
Package aggregator provides persistent storage and periodic snapshotting of aggregated analytics stats to PostgreSQL. |
|
Package collector provides a batch-oriented analytics event collector that accumulates events in memory and flushes them to Kafka in bulk.
|
Package collector provides a batch-oriented analytics event collector that accumulates events in memory and flushes them to Kafka in bulk. |