server

Documentation

Overview

Package server implements the gRPC service for time series data querying.

The server provides:

• Time series data querying with various aggregations
• Request validation and error handling
• Middleware support for:
• Request rate limiting
• Response caching
• Metrics collection
• Logging
• Context management
• Prometheus metrics integration
• gRPC reflection for debugging

Example Usage:

config := DefaultServerConfig()
repo := database.NewTimeScaleDB(...)

server, err := SetupServer(repo, config)
if err != nil {
    log.Fatalf("Failed to setup server: %v", err)
}

lis, err := net.Listen("tcp", ":50051")
if err != nil {
    log.Fatalf("Failed to listen: %v", err)
}

if err := server.Serve(lis); err != nil {
    log.Fatalf("Failed to serve: %v", err)
}

Index

• Constants
• func ConfigureGRPCServer(repo database.TimeSeriesRepository, opts ...grpc.ServerOption) *grpc.Server
• func SetupServer(repo database.TimeSeriesRepository, config ServerConfig) (*grpc.Server, error)
• func SetupServerWithRegistry(repo database.TimeSeriesRepository, logger *logrus.Logger, ...) (*grpc.Server, error)
• type HealthChecker
  • func NewHealthChecker() *HealthChecker
  • func (h *HealthChecker) Check(ctx context.Context, req *grpc_health_v1.HealthCheckRequest) (*grpc_health_v1.HealthCheckResponse, error)
  • func (h *HealthChecker) SetServingStatus(service string, status grpc_health_v1.HealthCheckResponse_ServingStatus)
  • func (h *HealthChecker) Watch(req *grpc_health_v1.HealthCheckRequest, ...) error
• type RequestValidator
  • func NewRequestValidator() *RequestValidator
  • func (v *RequestValidator) Validate(start, end time.Time, window, aggregation string) error
• type ServerConfig
  • func DefaultServerConfig() ServerConfig
• type TimeSeriesService
  • func NewTimeSeriesService(repo database.TimeSeriesRepository) *TimeSeriesService
  • func (s *TimeSeriesService) QueryTimeSeries(ctx context.Context, req *pb.TimeSeriesRequest) (*pb.TimeSeriesResponse, error)

Constants

const (
	Window1m = "1m"
	Window5m = "5m"
	Window1h = "1h"
	Window1d = "1d"

	AggregationMin = "MIN"
	AggregationMax = "MAX"
	AggregationAvg = "AVG"
	AggregationSum = "SUM"
)

Functions

func ConfigureGRPCServer

func ConfigureGRPCServer(
	repo database.TimeSeriesRepository,
	opts ...grpc.ServerOption,
) *grpc.Server

gRPC Server Configuration without the middleware (for development and debug only)

func SetupServer

func SetupServer(repo database.TimeSeriesRepository, config ServerConfig) (*grpc.Server, error)

SetupServer initializes and configures the gRPC server with all middleware

func SetupServerWithRegistry

func SetupServerWithRegistry(repo database.TimeSeriesRepository, logger *logrus.Logger, reg prometheus.Registerer) (*grpc.Server, error)

SetupServerWithRegistry initializes the server with a custom registry

Types

type HealthChecker

type HealthChecker struct {
	grpc_health_v1.UnimplementedHealthServer
	// contains filtered or unexported fields
}

HealthChecker implements the gRPC health checking protocol

func NewHealthChecker

func NewHealthChecker() *HealthChecker

func (*HealthChecker) Check

func (h *HealthChecker) Check(ctx context.Context, req *grpc_health_v1.HealthCheckRequest) (*grpc_health_v1.HealthCheckResponse, error)

func (*HealthChecker) SetServingStatus

func (h *HealthChecker) SetServingStatus(service string, status grpc_health_v1.HealthCheckResponse_ServingStatus)

SetServingStatus sets the serving status of a service

func (*HealthChecker) Watch

func (h *HealthChecker) Watch(req *grpc_health_v1.HealthCheckRequest, stream grpc_health_v1.Health_WatchServer) error

type RequestValidator

type RequestValidator struct {
	// contains filtered or unexported fields
}

func NewRequestValidator

func NewRequestValidator() *RequestValidator

func (*RequestValidator) Validate

func (v *RequestValidator) Validate(start, end time.Time, window, aggregation string) error

Validate checks if the request parameters are valid

type ServerConfig

type ServerConfig struct {
	CacheSize      int     // Size of the LRU cache
	RateLimit      float64 // Requests per second
	RateLimitBurst int     // Maximum burst size for rate limiting
}

ServerConfig holds configuration options for the gRPC server. It controls caching, rate limiting, and other server behaviors.

func DefaultServerConfig

func DefaultServerConfig() ServerConfig

DefaultServerConfig returns a ServerConfig with sensible defaults

type TimeSeriesService

type TimeSeriesService struct {
	pb.UnimplementedTimeSeriesServiceServer
	// contains filtered or unexported fields
}

TimeSeriesService implements the gRPC service for querying time series data. It handles request validation, data retrieval, and response formatting.

func NewTimeSeriesService

func NewTimeSeriesService(repo database.TimeSeriesRepository) *TimeSeriesService

NewTimeSeriesService creates a new service instance

func (*TimeSeriesService) QueryTimeSeries

func (s *TimeSeriesService) QueryTimeSeries(
	ctx context.Context,
	req *pb.TimeSeriesRequest,
) (*pb.TimeSeriesResponse, error)

QueryTimeSeries retrieves time series data based on the provided request parameters. It supports various time windows and aggregation methods.