proto

README

EdgeCom Energy Time Series Service

A gRPC service for fetching, storing, and querying time series data from EdgeCom Energy's API.

Features

• Historical data bootstrapping (up to 2 years)
• Time series data aggregation (MIN, MAX, AVG, SUM)
• Configurable time windows (1m, 5m, 1h, 1d)
• gRPC API with reflection support
• TimescaleDB integration for efficient time series storage
• Prometheus metrics integration
• Structured logging with logrus
• Request caching and rate limiting

Prerequisites

• Go 1.22.0 or later
• Docker and Docker Compose v2
• TimescaleDB 2.x
• Access to EdgeCom Energy API
• grpcurl (for testing)

Installation

1. Clone the repository:

git clone https://github.com/tejusbharadwaj/edgecom.git
cd edgecom

1. Copy the example config and modify as needed:

cp config.example.yaml config.yaml

1. Run all tests (unit tests and integration tests):

docker compose --profile test up --build

1. Run the application:

docker compose up --build

Service Ports

The service exposes:

• gRPC server on port 50051 (mapped from container port 8080)
• PostgreSQL/TimescaleDB on port 5432

Configuration

Configuration is managed through config.yaml:

server:
  port: 8080  # Container port (mapped to 50051 on host)
  host: "0.0.0.0"
  url: "https://api.edgecomenergy.net/core/asset/{asset-id}/series"

database:
  host: "db"
  port: 5432
  name: "edgecom"
  user: "edgecom"
  password: "edgecom"
  ssl_mode: "disable"
  max_connections: 10
  connection_timeout: 5

logging:
  level: "info"
  format: "json"

API Reference

gRPC Service Definition

service TimeSeriesService {
    rpc QueryTimeSeries(TimeSeriesRequest) returns (TimeSeriesResponse) {}
}

message TimeSeriesRequest {
    google.protobuf.Timestamp start = 1;
    google.protobuf.Timestamp end = 2;
    string window = 3;       // "1m", "5m", "1h", "1d"
    string aggregation = 4;  // "MIN", "MAX", "AVG", "SUM"
}

Testing the API

Using grpcurl:

# List available services
grpcurl -plaintext localhost:50051 list

# Query time series data
grpcurl -plaintext -d '{
  "start": "2024-11-23T00:00:00Z",
  "end": "2024-11-24T00:00:00Z",
  "window": "1h",
  "aggregation": "AVG"
}' localhost:50051 edgecom.TimeSeriesService/QueryTimeSeries

Development

Project Structure

.
├── cmd/                # Application entry point
├── internal/
│   ├── api/           # API client for EdgeCom Energy
│   ├── database/      # Database interactions
│   ├── grpc/          # gRPC service implementation
│   └── scheduler/     # Background job scheduler
├── k8s/               # Kubernetes manifests
├── proto/             # Protocol buffer definitions
├── scripts/           # Deployment scripts
│   ├── deploy.sh      # Kubernetes deployment script
│   └── cleanup.sh     # Kubernetes cleanup script
├── migrations/        # Database migrations
├── integration-tests/ # Integration tests
├── config.yaml       # Configuration file
└── docker-compose.yml

Running Tests

# Run unit tests
go test ./...

# Run integration tests
docker compose --profile test up --build

Building Locally

While you can build the application locally, it's recommended to use Docker Compose as it handles all configurations, dependencies, and environment setup automatically.

Option 1: Using Docker Compose (Recommended)

# This will handle all configurations, database setup, and dependencies
docker compose up --build

Option 2: Manual Build (Advanced)

# Only use this if you have specific requirements that prevent using Docker Compose
# You'll need to:
# 1. Set up TimescaleDB manually
# 2. Configure environment variables
# 3. Handle dependencies

go build -o edgecom ./cmd/main.go

Note: Docker Compose is the preferred method as it ensures consistent environments and handles all necessary configurations. Only use manual building if you have specific requirements that prevent using Docker Compose.

Monitoring

The service includes:

• Request rate limiting (5 req/s with burst of 10)
• LRU cache for frequent queries (1000 entries)
• Prometheus metrics for:
  • Request counts
  • Request latencies
  • Cache hit/miss ratios

Error Handling

The service implements graceful degradation:

• Validates all incoming requests
• Implements retry logic for API requests
• Provides detailed error logging
• Graceful shutdown handling

Deployment Options

Option 1: Docker Compose (Primary Method)

Docker Compose is the preferred and officially supported deployment method for this service. It was designed and optimized for Docker Compose deployment, making it the most reliable and straightforward option.

# Start the service
docker compose up --build

# Stop the service
docker compose down

For development and testing:

# Run all tests (unit tests and integration tests)
docker compose --profile test up --build

Option 2: Kubernetes (Alternative)

Note: While Kubernetes deployment is supported, Docker Compose is the primary and recommended method. Use Kubernetes only if it's specifically required for your infrastructure needs.

Prerequisites for Kubernetes Deployment

• Kubernetes cluster (local or cloud)
• kubectl configured with your cluster
• Helm (optional, for database deployment)

Kubernetes Deployment Steps

1. Create required Kubernetes resources:

# Using deployment script
./scripts/deploy.sh

# Or manually apply each manifest
kubectl apply -f k8s/config.yaml
kubectl apply -f k8s/database.yaml
kubectl apply -f k8s/database-service.yaml
kubectl apply -f k8s/deployment.yaml
kubectl apply -f k8s/service.yaml

1. Verify the deployment:

kubectl get pods
kubectl get services

1. To cleanup Kubernetes resources:

./scripts/cleanup.sh

Deployment Environments

Local Development (Recommended)

• Use Docker Compose for the simplest and most reliable setup
• Automatic hot-reloading for development
• Integrated test environment
• Matches the primary deployment method
• Minimal configuration required

Production

Using Docker Compose (Recommended)

• Simple, reliable deployment
• Easy configuration management
• Straightforward scaling
• Built-in service discovery
• Automatic container recovery

Using Kubernetes (Alternative)

• Available for specific infrastructure requirements
• Scalable deployment with Kubernetes
• Configurable resource limits
• Rolling updates support
• Health checks and auto-healing
• Load balancing

Documentation

Index

• Constants
• Variables
• func RegisterTimeSeriesServiceServer(s grpc.ServiceRegistrar, srv TimeSeriesServiceServer)
• type TimeSeriesDataPoint
  • func (*TimeSeriesDataPoint) Descriptor() ([]byte, []int)
  • func (x *TimeSeriesDataPoint) GetTime() *timestamppb.Timestamp
  • func (x *TimeSeriesDataPoint) GetValue() float64
  • func (*TimeSeriesDataPoint) ProtoMessage()
  • func (x *TimeSeriesDataPoint) ProtoReflect() protoreflect.Message
  • func (x *TimeSeriesDataPoint) Reset()
  • func (x *TimeSeriesDataPoint) String() string
• type TimeSeriesRequest
  • func (*TimeSeriesRequest) Descriptor() ([]byte, []int)
  • func (x *TimeSeriesRequest) GetAggregation() string
  • func (x *TimeSeriesRequest) GetEnd() *timestamppb.Timestamp
  • func (x *TimeSeriesRequest) GetStart() *timestamppb.Timestamp
  • func (x *TimeSeriesRequest) GetWindow() string
  • func (*TimeSeriesRequest) ProtoMessage()
  • func (x *TimeSeriesRequest) ProtoReflect() protoreflect.Message
  • func (x *TimeSeriesRequest) Reset()
  • func (x *TimeSeriesRequest) String() string
• type TimeSeriesResponse
  • func (*TimeSeriesResponse) Descriptor() ([]byte, []int)
  • func (x *TimeSeriesResponse) GetData() []*TimeSeriesDataPoint
  • func (*TimeSeriesResponse) ProtoMessage()
  • func (x *TimeSeriesResponse) ProtoReflect() protoreflect.Message
  • func (x *TimeSeriesResponse) Reset()
  • func (x *TimeSeriesResponse) String() string
• type TimeSeriesServiceClient
  • func NewTimeSeriesServiceClient(cc grpc.ClientConnInterface) TimeSeriesServiceClient
• type TimeSeriesServiceServer
• type UnimplementedTimeSeriesServiceServer
  • func (UnimplementedTimeSeriesServiceServer) QueryTimeSeries(context.Context, *TimeSeriesRequest) (*TimeSeriesResponse, error)
• type UnsafeTimeSeriesServiceServer

Constants

const (
	TimeSeriesService_QueryTimeSeries_FullMethodName = "/edgecom.TimeSeriesService/QueryTimeSeries"
)

Variables

var File_timeseries_proto protoreflect.FileDescriptor

var TimeSeriesService_ServiceDesc = grpc.ServiceDesc{
	ServiceName: "edgecom.TimeSeriesService",
	HandlerType: (*TimeSeriesServiceServer)(nil),
	Methods: []grpc.MethodDesc{
		{
			MethodName: "QueryTimeSeries",
			Handler:    _TimeSeriesService_QueryTimeSeries_Handler,
		},
	},
	Streams:  []grpc.StreamDesc{},
	Metadata: "timeseries.proto",
}

TimeSeriesService_ServiceDesc is the grpc.ServiceDesc for TimeSeriesService service. It's only intended for direct use with grpc.RegisterService, and not to be introspected or modified (even as a copy)

Functions

func RegisterTimeSeriesServiceServer

func RegisterTimeSeriesServiceServer(s grpc.ServiceRegistrar, srv TimeSeriesServiceServer)

Types

type TimeSeriesDataPoint

type TimeSeriesDataPoint struct {
	Time  *timestamppb.Timestamp `protobuf:"bytes,1,opt,name=time,proto3" json:"time,omitempty"`
	Value float64                `protobuf:"fixed64,2,opt,name=value,proto3" json:"value,omitempty"`
	// contains filtered or unexported fields
}

func (*TimeSeriesDataPoint) Descriptor

func (*TimeSeriesDataPoint) Descriptor() ([]byte, []int)

Deprecated: Use TimeSeriesDataPoint.ProtoReflect.Descriptor instead.

func (*TimeSeriesDataPoint) GetTime

func (x *TimeSeriesDataPoint) GetTime() *timestamppb.Timestamp

func (*TimeSeriesDataPoint) GetValue

func (x *TimeSeriesDataPoint) GetValue() float64

func (*TimeSeriesDataPoint) ProtoMessage

func (*TimeSeriesDataPoint) ProtoMessage()

func (*TimeSeriesDataPoint) ProtoReflect

func (x *TimeSeriesDataPoint) ProtoReflect() protoreflect.Message

func (*TimeSeriesDataPoint) Reset

func (x *TimeSeriesDataPoint) Reset()

func (*TimeSeriesDataPoint) String

func (x *TimeSeriesDataPoint) String() string

type TimeSeriesRequest

type TimeSeriesRequest struct {
	Start       *timestamppb.Timestamp `protobuf:"bytes,1,opt,name=start,proto3" json:"start,omitempty"`
	End         *timestamppb.Timestamp `protobuf:"bytes,2,opt,name=end,proto3" json:"end,omitempty"`
	Window      string                 `protobuf:"bytes,3,opt,name=window,proto3" json:"window,omitempty"`           // e.g., '1m', '5m', '1h', '1d'
	Aggregation string                 `protobuf:"bytes,4,opt,name=aggregation,proto3" json:"aggregation,omitempty"` // 'MIN', 'MAX', 'AVG', 'SUM'
	// contains filtered or unexported fields
}

func (*TimeSeriesRequest) Descriptor

func (*TimeSeriesRequest) Descriptor() ([]byte, []int)

Deprecated: Use TimeSeriesRequest.ProtoReflect.Descriptor instead.

func (*TimeSeriesRequest) GetAggregation

func (x *TimeSeriesRequest) GetAggregation() string

func (*TimeSeriesRequest) GetEnd

func (x *TimeSeriesRequest) GetEnd() *timestamppb.Timestamp

func (*TimeSeriesRequest) GetStart

func (x *TimeSeriesRequest) GetStart() *timestamppb.Timestamp

func (*TimeSeriesRequest) GetWindow

func (x *TimeSeriesRequest) GetWindow() string

func (*TimeSeriesRequest) ProtoMessage

func (*TimeSeriesRequest) ProtoMessage()

func (*TimeSeriesRequest) ProtoReflect

func (x *TimeSeriesRequest) ProtoReflect() protoreflect.Message

func (*TimeSeriesRequest) Reset

func (x *TimeSeriesRequest) Reset()

func (*TimeSeriesRequest) String

func (x *TimeSeriesRequest) String() string

type TimeSeriesResponse

type TimeSeriesResponse struct {
	Data []*TimeSeriesDataPoint `protobuf:"bytes,1,rep,name=data,proto3" json:"data,omitempty"`
	// contains filtered or unexported fields
}

func (*TimeSeriesResponse) Descriptor

func (*TimeSeriesResponse) Descriptor() ([]byte, []int)

Deprecated: Use TimeSeriesResponse.ProtoReflect.Descriptor instead.

func (*TimeSeriesResponse) GetData

func (x *TimeSeriesResponse) GetData() []*TimeSeriesDataPoint

func (*TimeSeriesResponse) ProtoMessage

func (*TimeSeriesResponse) ProtoMessage()

func (*TimeSeriesResponse) ProtoReflect

func (x *TimeSeriesResponse) ProtoReflect() protoreflect.Message

func (*TimeSeriesResponse) Reset

func (x *TimeSeriesResponse) Reset()

func (*TimeSeriesResponse) String

func (x *TimeSeriesResponse) String() string

type TimeSeriesServiceClient

type TimeSeriesServiceClient interface {
	QueryTimeSeries(ctx context.Context, in *TimeSeriesRequest, opts ...grpc.CallOption) (*TimeSeriesResponse, error)
}

TimeSeriesServiceClient is the client API for TimeSeriesService 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.

func NewTimeSeriesServiceClient

func NewTimeSeriesServiceClient(cc grpc.ClientConnInterface) TimeSeriesServiceClient

type TimeSeriesServiceServer

type TimeSeriesServiceServer interface {
	QueryTimeSeries(context.Context, *TimeSeriesRequest) (*TimeSeriesResponse, error)
	// contains filtered or unexported methods
}

TimeSeriesServiceServer is the server API for TimeSeriesService service. All implementations must embed UnimplementedTimeSeriesServiceServer for forward compatibility.

type UnimplementedTimeSeriesServiceServer

type UnimplementedTimeSeriesServiceServer struct{}

UnimplementedTimeSeriesServiceServer 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.

func (UnimplementedTimeSeriesServiceServer) QueryTimeSeries

func (UnimplementedTimeSeriesServiceServer) QueryTimeSeries(context.Context, *TimeSeriesRequest) (*TimeSeriesResponse, error)

type UnsafeTimeSeriesServiceServer

type UnsafeTimeSeriesServiceServer interface {
	// contains filtered or unexported methods
}

UnsafeTimeSeriesServiceServer may be embedded to opt out of forward compatibility for this service. Use of this interface is not recommended, as added methods to TimeSeriesServiceServer will result in compilation errors.