Merge branch 'feat/redis-current-store'

Author: NeedleInAJayStack

README.md

@@ -4,21 +4,30 @@ This is Jay's Timeseries API server. It requires a Timescale DB instance.
 
 ## Getting Started
 
-Create a `.env` file in this directory that contains necessary environment variables. For example:
+Create a `.env` file in this directory that contains necessary environment variables. Defaults:
 
 ```
-USERNAME=<username>
-PASSWORD=<password>
-API_KEY=<apikey>
-
-HOST=<host>
-PORT=<port>
-JWT_SECRET=<password>
-
-DATABASE_HOST=<host>
-DATABASE_PORT=<port>
-DATABASE_USERNAME=<username>
-DATABASE_PASSWORD=<password>
-DATABASE_NAME=<name>
+USERNAME=
+PASSWORD=
+API_KEY=
+
+HOST=localhost
+PORT=80
+JWT_SECRET=
+
+DATABASE_HOST=localhost
+DATABASE_PORT=5432
+DATABASE_USERNAME=postgres
+DATABASE_PASSWORD=postgres
+DATABASE_NAME=postgres
+
+CURRENT_STORE_TYPE=memory # Options: redis, memory
+REDIS_ADDRESS=localhost:6379
+REDIS_PASSWORD=
+REDIS_DATABASE=0
+
+MQTT_ADDRESS=
+MQTT_USERNAME=
+MQTT_PASSWORD=
 ```
 

currentController.go

@@ -29,7 +29,12 @@ func (h currentController) getCurrent(w http.ResponseWriter, request *http.Reque
 		return
 	}
 
-	httpResult := h.store.getCurrent(pointId)
+	httpResult, err := h.store.getCurrent(pointId)
+	if err != nil {
+		log.Printf("Cannot retrieve current value")
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
 
 	httpJson, err := json.Marshal(httpResult)
 	if err != nil {
@@ -61,7 +66,12 @@ func (h currentController) postCurrent(writer http.ResponseWriter, request *http
 		return
 	}
 
-	h.store.setCurrent(pointId, currentItem)
+	err = h.store.setCurrent(pointId, currentItem)
+	if err != nil {
+		log.Printf("Cannot save current value")
+		writer.WriteHeader(http.StatusInternalServerError)
+		return
+	}
 
 	writer.WriteHeader(http.StatusOK)
 }

currentStore.go

@@ -1,16 +1,20 @@
 package main
 
 import (
+	"context"
+	"encoding/json"
+	"log"
 	"sync"
 	"time"
 
 	"github.com/google/uuid"
+	"github.com/redis/go-redis/v9"
 )
 
 // currentStore is able to store point current values
 type currentStore interface {
-	getCurrent(uuid.UUID) current
-	setCurrent(uuid.UUID, currentInput)
+	getCurrent(uuid.UUID) (current, error)
+	setCurrent(uuid.UUID, currentInput) error
 }
 
 type currentInput struct {
@@ -36,18 +40,67 @@ func newInMemoryCurrentStore() inMemoryCurrentStore {
 	}
 }
 
-func (s inMemoryCurrentStore) getCurrent(id uuid.UUID) current {
+func (s inMemoryCurrentStore) getCurrent(id uuid.UUID) (current, error) {
 	s.mux.Lock()
 	defer s.mux.Unlock()
-	return s.cache[id]
+	return s.cache[id], nil
 }
 
-func (s inMemoryCurrentStore) setCurrent(id uuid.UUID, input currentInput) {
+func (s inMemoryCurrentStore) setCurrent(id uuid.UUID, input currentInput) error {
 	timestamp := time.Now()
 	s.mux.Lock()
 	defer s.mux.Unlock()
 	s.cache[id] = current{
 		Ts:    &timestamp,
 		Value: input.Value,
 	}
+	return nil
+}
+
+// redisCurrentStore stores point current values in a Redis database.
+type redisCurrentStore struct {
+	db        *redis.Client
+	keyPrefix string
+	ctx       context.Context
+}
+
+func newRedisCurrentStore(db *redis.Client) redisCurrentStore {
+	return redisCurrentStore{
+		db:        db,
+		keyPrefix: "timeseries-api:",
+		ctx:       context.Background(),
+	}
+}
+
+func (s redisCurrentStore) getCurrent(id uuid.UUID) (current, error) {
+	currentJson, err := s.db.Get(s.ctx, s.keyPrefix+id.String()).Bytes()
+	if err != nil {
+		log.Printf("Cannot retrieve current: %s", err)
+		return current{}, err
+	}
+	var result current
+	err = json.Unmarshal(currentJson, &result)
+	if err != nil {
+		log.Printf("Cannot decode current JSON: %s", err)
+		return current{}, err
+	}
+	return result, nil
+}
+
+func (s redisCurrentStore) setCurrent(id uuid.UUID, input currentInput) error {
+	timestamp := time.Now()
+	currentJson, err := json.Marshal(current{
+		Ts:    &timestamp,
+		Value: input.Value,
+	})
+	if err != nil {
+		log.Printf("Cannot encode current JSON: %s", err)
+		return err
+	}
+	err = s.db.Set(s.ctx, s.keyPrefix+id.String(), currentJson, 0).Err()
+	if err != nil {
+		log.Printf("Cannot store current: %s", err)
+		return err
+	}
+	return nil
 }

docker-compose.yaml

@@ -1,20 +1,22 @@
 services:
   timeseries-api:
+    profiles:
+      - local
     build:
       context: .
       dockerfile: Dockerfile
     env_file:
       - .env
     environment:
-      - DATABASE_HOST=db
+      - DATABASE_HOST=postgres
     ports:
       - '8080:80'
       - '2112:2112' # Metrics
     volumes:
       - .env:/project/.env
     depends_on:
-      - db
-  db:
+      - postgres
+  postgres:
     image: timescale/timescaledb:latest-pg16
     restart: unless-stopped
     volumes:
@@ -26,6 +28,11 @@ services:
       POSTGRES_PASSWORD: postgres
     ports:
       - '${DATABASE_PORT:-5432}:5432'
+  redis:
+    image: redis:latest
+    restart: unless-stopped
+    ports:
+      - '${REDIS_PORT:-6379}:6379'
 
 volumes:
   db_data:

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/joho/godotenv v1.5.1
 	github.com/prometheus/client_golang v1.20.5
+	github.com/redis/go-redis/v9 v9.7.0
 	github.com/stretchr/testify v1.9.0
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.57.0
 	go.opentelemetry.io/otel v1.32.0
@@ -19,6 +20,7 @@ require (
 )
 
 require (
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
@@ -49,12 +51,12 @@ require (
 	github.com/prometheus/common v0.60.1 // indirect
 	github.com/prometheus/procfs v0.15.1 // indirect
 	go.opentelemetry.io/otel/exporters/prometheus v0.54.0
-	golang.org/x/crypto v0.27.0 // indirect
-	golang.org/x/net v0.29.0 // indirect
-	golang.org/x/sync v0.8.0 // indirect
-	golang.org/x/sys v0.27.0 // indirect
-	golang.org/x/text v0.18.0 // indirect
-	google.golang.org/protobuf v1.35.1 // indirect
+	golang.org/x/crypto v0.31.0 // indirect
+	golang.org/x/net v0.33.0 // indirect
+	golang.org/x/sync v0.10.0 // indirect
+	golang.org/x/sys v0.28.0 // indirect
+	golang.org/x/text v0.21.0 // indirect
+	google.golang.org/protobuf v1.36.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	gorm.io/driver/mysql v1.5.6 // indirect
 )

go.sum

@@ -2,11 +2,17 @@ filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA=
 filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
+github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
+github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
+github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
+github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
 github.com/eclipse/paho.mqtt.golang v1.5.0 h1:EH+bUVJNgttidWFkLLVKaQPGmkTUfQQqjOsyvMGvD6o=
 github.com/eclipse/paho.mqtt.golang v1.5.0/go.mod h1:du/2qNQVqJf/Sqs4MEL77kR8QTqANF7XU7Fk0aOTAgk=
 github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg=
@@ -69,6 +75,8 @@ github.com/prometheus/common v0.60.1 h1:FUas6GcOw66yB/73KC+BOZoFJmbo/1pojoILArPA
 github.com/prometheus/common v0.60.1/go.mod h1:h0LYf1R1deLSKtD4Vdg8gy4RuOvENW2J/h19V5NADQw=
 github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0learggepc=
 github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk=
+github.com/redis/go-redis/v9 v9.7.0 h1:HhLSs+B6O021gwzl+locl0zEDnyNkxMtf/Z3NNBMa9E=
+github.com/redis/go-redis/v9 v9.7.0/go.mod h1:f6zhXITC7JUJIlPEiBOTXxJgPLdZcA93GewI7inzyWw=
 github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8=
 github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
@@ -90,18 +98,18 @@ go.opentelemetry.io/otel/sdk/metric v1.32.0 h1:rZvFnvmvawYb0alrYkjraqJq0Z4ZUJAiy
 go.opentelemetry.io/otel/sdk/metric v1.32.0/go.mod h1:PWeZlq0zt9YkYAp3gjKZ0eicRYvOh1Gd+X99x6GHpCQ=
 go.opentelemetry.io/otel/trace v1.32.0 h1:WIC9mYrXf8TmY/EXuULKc8hR17vE+Hjv2cssQDe03fM=
 go.opentelemetry.io/otel/trace v1.32.0/go.mod h1:+i4rkvCraA+tG6AzwloGaCtkx53Fa+L+V8e9a7YvhT8=
-golang.org/x/crypto v0.27.0 h1:GXm2NjJrPaiv/h1tb2UH8QfgC/hOf/+z0p6PT8o1w7A=
-golang.org/x/crypto v0.27.0/go.mod h1:1Xngt8kV6Dvbssa53Ziq6Eqn0HqbZi5Z6R0ZpwQzt70=
-golang.org/x/net v0.29.0 h1:5ORfpBpCs4HzDYoodCDBbwHzdR5UrLBZ3sOnUJmFoHo=
-golang.org/x/net v0.29.0/go.mod h1:gLkgy8jTGERgjzMic6DS9+SP0ajcu6Xu3Orq/SpETg0=
-golang.org/x/sync v0.8.0 h1:3NFvSEYkUoMifnESzZl15y791HH1qU2xm6eCJU5ZPXQ=
-golang.org/x/sync v0.8.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
-golang.org/x/sys v0.27.0 h1:wBqf8DvsY9Y/2P8gAfPDEYNuS30J4lPHJxXSb/nJZ+s=
-golang.org/x/sys v0.27.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
-golang.org/x/text v0.18.0 h1:XvMDiNzPAl0jr17s6W9lcaIhGUfUORdGCNsuLmPG224=
-golang.org/x/text v0.18.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY=
-google.golang.org/protobuf v1.35.1 h1:m3LfL6/Ca+fqnjnlqQXNpFPABW1UD7mjh8KO2mKFytA=
-google.golang.org/protobuf v1.35.1/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
+golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U=
+golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk=
+golang.org/x/net v0.33.0 h1:74SYHlV8BIgHIFC/LrYkOGIwL19eTYXQ5wc6TBuO36I=
+golang.org/x/net v0.33.0/go.mod h1:HXLR5J+9DxmrqMwG9qjGCxZ+zKXxBru04zlTvWlWuN4=
+golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ=
+golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA=
+golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
+golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
+google.golang.org/protobuf v1.36.1 h1:yBPeRvTftaleIgM3PZ/WBIZ7XM/eEYAaEyCwvyjq/gk=
+google.golang.org/protobuf v1.36.1/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=

ingester_test.go

@@ -65,9 +65,12 @@ func (suite *IngesterTestSuite) TestIngester() {
 	assert.Equal(suite.T(), suite.ingester.topics["test"][rec1.ID], true)
 	assert.Equal(suite.T(), suite.ingester.topics["test"][rec2.ID], true)
 	assert.Equal(suite.T(), suite.ingester.topics["test2"][rec3.ID], true)
-	assert.Equal(suite.T(), *suite.currentStore.getCurrent(rec1.ID).Value, 0.0)
-	assert.Equal(suite.T(), *suite.currentStore.getCurrent(rec2.ID).Value, 0.0)
-	assert.Equal(suite.T(), *suite.currentStore.getCurrent(rec3.ID).Value, 0.0)
+	actualRec1, _ := suite.currentStore.getCurrent(rec1.ID)
+	assert.Equal(suite.T(), *actualRec1.Value, 0.0)
+	actualRec2, _ := suite.currentStore.getCurrent(rec2.ID)
+	assert.Equal(suite.T(), *actualRec2.Value, 0.0)
+	actualRec3, _ := suite.currentStore.getCurrent(rec3.ID)
+	assert.Equal(suite.T(), *actualRec3.Value, 0.0)
 
 	// Check that removing some records removes subscriptions and emit no longer sets their current value
 	suite.ingester.refreshSubscriptions(
@@ -79,7 +82,10 @@ func (suite *IngesterTestSuite) TestIngester() {
 	assert.Equal(suite.T(), suite.ingester.topics["test"][rec1.ID], true)
 	assert.Equal(suite.T(), suite.ingester.topics["test"][rec2.ID], false)
 	assert.Equal(suite.T(), suite.ingester.topics["test2"][rec3.ID], false)
-	assert.Equal(suite.T(), *suite.currentStore.getCurrent(rec1.ID).Value, 1.0)
-	assert.Equal(suite.T(), *suite.currentStore.getCurrent(rec2.ID).Value, 0.0)
-	assert.Equal(suite.T(), *suite.currentStore.getCurrent(rec3.ID).Value, 0.0)
+	actualRec1, _ = suite.currentStore.getCurrent(rec1.ID)
+	assert.Equal(suite.T(), *actualRec1.Value, 1.0)
+	actualRec2, _ = suite.currentStore.getCurrent(rec2.ID)
+	assert.Equal(suite.T(), *actualRec2.Value, 0.0)
+	actualRec3, _ = suite.currentStore.getCurrent(rec3.ID)
+	assert.Equal(suite.T(), *actualRec3.Value, 0.0)
 }

main.go

@@ -11,6 +11,7 @@ import (
 	mqtt "github.com/eclipse/paho.mqtt.golang"
 	"github.com/joho/godotenv"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"github.com/redis/go-redis/v9"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/exporters/prometheus"
 	"go.opentelemetry.io/otel/sdk/metric"
@@ -70,7 +71,25 @@ func main() {
 	}
 	historyStore := newGormHistoryStore(db)
 	recStore := newGormRecStore(db)
-	currentStore := newInMemoryCurrentStore()
+
+	var currentStore currentStore
+	currentStoreType := envOrDefault("CURRENT_STORE_TYPE", "memory")
+	if currentStoreType == "redis" {
+		redisDatabase, err := strconv.Atoi(envOrDefault("REDIS_DATABASE", "0"))
+		if err != nil {
+			log.Fatal(err)
+		}
+		redisClient := redis.NewClient(&redis.Options{
+			Addr:     envOrDefault("REDIS_ADDRESS", "localhost:6379"),
+			Password: envOrDefault("REDIS_PASSWORD", ""),
+			DB:       redisDatabase,
+		})
+		currentStore = newRedisCurrentStore(redisClient)
+	} else if currentStoreType == "memory" {
+		currentStore = newInMemoryCurrentStore()
+	} else {
+		log.Fatalf("Unknown current store type: %s", currentStoreType)
+	}
 
 	serverConfig := ServerConfig{
 		authenticator:        authenticator,