fix recorder npe, implement filters

.gitea/workflows/ci.yml

@@ -2,6 +2,7 @@ name: Build and Publish
 on:
   push:
     tags: ["v*"]
+    branches: ["main"]
 
 env:
   PACKAGE_NAME: ambient-local-exporter
@@ -21,9 +22,10 @@ env:
 jobs:
   release:
     runs-on: ubuntu-latest
+    if: startsWith(github.ref, 'refs/tags/v')  # Only run on tag push
     steps:
       - name: Checkout Code
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
 
       - name: Set up Go Environment
         uses: actions/setup-go@v4
@@ -87,12 +89,29 @@ jobs:
             VER_PKG=${{ env.VER_PKG }}
             VERSION=${{ github.ref_name }}
 
+  # Detect if the helm chart was updated
+  check-chart:
+    runs-on: ubuntu-latest
+    outputs:
+      chart-updated: ${{ steps.filter.outputs.chart }}
+    steps:
+      - uses: actions/checkout@v4
+      - name: Check Chart Changed
+        uses: dorny/paths-filter@v3
+        id: filter
+        with:
+          base: ${{ github.ref }}
+          filters: |
+            chart:
+              - helm/ambient-local-exporter/Chart.yaml
+
   helm-release:
     runs-on: ubuntu-latest
-    needs: release
+    needs: check-chart
+    if: ${{ needs.check-chart.outputs.chart-updated == 'true' }}
     steps:
       - name: Checkout Code
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
 
       - name: Install Helm
         env:
@@ -102,7 +121,7 @@ jobs:
 
       - name: Package Chart
         run: |
-          helm package --app-version ${VERSION#v} ${CHART_DIR}
+          helm package --app-version ${VERSION} ${CHART_DIR}
 
       - name: Publish Chart
         env:

docker-compose-sample.yml

@@ -3,7 +3,8 @@ services:
   ambient-local-exporter:
     build: .
     ports:
-      - 8080:8080
+      - 8080:8080 # HTTP
+      - 8081:8081 # GRPC
     volumes:
       - ./config.yaml:/app/config.yaml
     command:

helm/ambient-local-exporter/Chart.yaml

@@ -15,13 +15,13 @@ type: application
 # This is the chart version. This version number should be incremented each time you make changes
 # to the chart and its templates, including the app version.
 # Versions are expected to follow Semantic Versioning (https://semver.org/)
-version: 0.1.0
+version: 0.1.2
 
 # This is the version number of the application being deployed. This version number should be
 # incremented each time you make changes to the application. Versions are not expected to
 # follow Semantic Versioning. They should reflect the version the application is using.
 # It is recommended to use it with quotes.
-appVersion: "0.9.0"
+appVersion: "v0.10.2"
 
 dependencies:
   - name: hull

helm/ambient-local-exporter/values.yaml

@@ -130,7 +130,7 @@ hull:
             main:
               image:
                 repository: _HT*hull.config.settings.repo
-                tag: _HT!{{ printf "v%s" _HT*hull.config.settings.tag }}
+                tag: _HT*hull.config.settings.tag
               imagePullPolicy: Always
               args:
                 - -config

main.go

@@ -9,19 +9,16 @@ import (
 	grpcopts "gitea.libretechconsulting.com/rmcguire/go-app/pkg/srv/grpc/opts"
 	httpopts "gitea.libretechconsulting.com/rmcguire/go-app/pkg/srv/http/opts"
 	"golang.org/x/sys/unix"
-	"k8s.io/utils/ptr"
 
 	weatherpb "gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/api/v1alpha1/weather"
 	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/ambient"
 	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/ambient/ambienthttp"
 	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/ambient/config"
 	weathergrpc "gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/weather/grpc"
-	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/weather/state"
 )
 
 const (
-	defaultMetricPrefix  = "weather"
-	defaultUpdatesToKeep = 120
+	defaultMetricPrefix = "weather"
 )
 
 func main() {
@@ -48,11 +45,6 @@ func main() {
 }
 
 func prepareApp(ctx context.Context, aw *ambient.AmbientWeather) *app.App {
-	// Config updates / defaults
-	if aw.Config.UpdatesToKeep == nil || *aw.Config.UpdatesToKeep < 1 {
-		aw.Config.UpdatesToKeep = ptr.To(defaultUpdatesToKeep)
-	}
-
 	// Load ambient routes into app
 	awApp := &app.App{
 		AppContext: ctx,
@@ -88,13 +80,9 @@ func prepareApp(ctx context.Context, aw *ambient.AmbientWeather) *app.App {
 		GRPC: &grpcopts.AppGRPC{
 			Services: []*grpcopts.GRPCService{
 				{
-					Name: "Weather Service",
-					Type: &weatherpb.AmbientLocalWeatherService_ServiceDesc,
-					Service: weathergrpc.NewGRPCWeather(ctx, state.NewWeatherState(
-						&state.Opts{
-							Ctx:      ctx,
-							KeepLast: *aw.Config.UpdatesToKeep,
-						})),
+					Name:    "Weather Service",
+					Type:    &weatherpb.AmbientLocalWeatherService_ServiceDesc,
+					Service: weathergrpc.NewGRPCWeather(ctx, aw.GetRecorder()),
 				},
 			},
 		},

pkg/ambient/ambient.go

@@ -20,35 +20,64 @@ import (
 	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/provider/awn"
 	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/provider/wunderground"
 	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/weather"
+	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/weather/recorder"
 )
 
+const defUpdatesToKeep = 120
+
 type AmbientWeather struct {
 	// These providers implement support for the update sent
 	// when either "AmbientWeather" or "Wunderground" are selected
 	// in the "Custom" section of the AWNet app, or the web UI
 	// of an Ambient WeatherHub
-	Config      *config.AmbientLocalExporterConfig
-	awnProvider provider.AmbientProvider
-	wuProvider  provider.AmbientProvider
-	appCtx      context.Context
-	metrics     *weather.WeatherMetrics
-	l           *zerolog.Logger
+	Config          *config.AmbientLocalExporterConfig
+	awnProvider     provider.AmbientProvider
+	wuProvider      provider.AmbientProvider
+	weatherRecorder *recorder.WeatherRecorder
+	appCtx          context.Context
+	metrics         *weather.WeatherMetrics
+	l               *zerolog.Logger
+	*sync.RWMutex
 }
 
 func New(appCtx context.Context, awConfig *config.AmbientLocalExporterConfig) *AmbientWeather {
 	return &AmbientWeather{
-		Config: awConfig,
-		appCtx: appCtx,
+		Config:  awConfig,
+		appCtx:  appCtx,
+		RWMutex: &sync.RWMutex{},
 	}
 }
 
 // Initialize with defaults, set logger from context
 func (aw *AmbientWeather) Init() *AmbientWeather {
+	tracer := otel.GetTracer(aw.appCtx, "ambientWeather")
+	_, span := tracer.Start(aw.appCtx, "ambientWeather.init",
+		trace.WithAttributes(
+			attribute.String("name", aw.Config.Name),
+			attribute.Bool("grpcEnabled", aw.Config.GRPCEnabled()),
+			attribute.Bool("httpEnabled", aw.Config.HTTPEnabled()),
+			attribute.Int("weatherStations", len(aw.Config.WeatherStations)),
+		))
+	defer span.End()
+
 	aw.awnProvider = &awn.AWNProvider{}
 	aw.wuProvider = &wunderground.WUProvider{}
 	aw.l = zerolog.Ctx(aw.appCtx)
 
+	updatesToKeep := defUpdatesToKeep
+	if aw.Config.UpdatesToKeep != nil && *aw.Config.UpdatesToKeep > 0 {
+		updatesToKeep = *aw.Config.UpdatesToKeep
+	}
+	span.SetAttributes(attribute.Int("updatesToKeep", updatesToKeep))
+
+	// TODO: Support other recorders (don't rely on default)
+	aw.weatherRecorder = recorder.NewWeatherRecorder(&recorder.Opts{
+		Ctx:      aw.appCtx,
+		KeepLast: updatesToKeep,
+	})
+
 	aw.l.Trace().Any("awConfig", aw.Config).Send()
+	span.SetStatus(codes.Ok, "")
 	return aw
 }
 
@@ -78,6 +107,9 @@ func (aw *AmbientWeather) handleProviderRequest(
 	w http.ResponseWriter,
 	r *http.Request,
 ) {
+	aw.Lock()
+	defer aw.Unlock()
+
 	l := zerolog.Ctx(aw.appCtx)
 	tracer := otel.GetTracer(aw.appCtx, p.Name()+".http.handler")
 
@@ -110,6 +142,9 @@ func (aw *AmbientWeather) handleProviderRequest(
 		updateSpan.SetAttributes(attribute.String("stationName", update.StationConfig.Name))
 	}
 
+	// Record state
+	aw.weatherRecorder.Set(ctx, update)
+
 	// Update metrics
 	aw.metricsUpdate(ctx, p, update)
 
@@ -242,3 +277,10 @@ func (aw *AmbientWeather) enrichStation(update *weather.WeatherUpdate) {
 		}
 	}
 }
+
+func (aw *AmbientWeather) GetRecorder() *recorder.WeatherRecorder {
+	aw.RLock()
+	defer aw.RUnlock()
+
+	return aw.weatherRecorder
+}

pkg/util/util.go

@@ -0,0 +1,17 @@
+package util
+
+import "k8s.io/utils/ptr"
+
+func DerefStr(s *string) string {
+	if s == nil {
+		return ""
+	}
+	return *s
+}
+
+func Int32ptr(i *int) *int32 {
+	if i == nil {
+		return nil
+	}
+	return ptr.To(int32(*i))
+}

pkg/weather/grpc/mapupdate.go

@@ -1,9 +1,8 @@
 package grpc
 
 import (
-	"k8s.io/utils/ptr"
-
 	pb "gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/api/v1alpha1/weather"
+	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/util"
 	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/weather"
 )
 
@@ -18,18 +17,18 @@ func UpdatesToPbUpdates(u []*weather.WeatherUpdate) []*pb.WeatherUpdate {
 func UpdateToPbUpdate(u *weather.WeatherUpdate) *pb.WeatherUpdate {
 	return &pb.WeatherUpdate{
 		StationName:         u.StationConfig.Name,
-		StationType:         derefStr(u.StationType),
-		StationId:           derefStr(u.StationID),
+		StationType:         util.DerefStr(u.StationType),
+		StationId:           util.DerefStr(u.StationID),
 		TempOutdoorF:        u.TempOutdoorF,
 		TempIndoorF:         u.TempIndoorF,
-		HumidityOutdoor:     int32ptr(u.HumidityOudoor),
-		HumidityIndoor:      int32ptr(u.HumidityIndoor),
+		HumidityOutdoor:     util.Int32ptr(u.HumidityOudoor),
+		HumidityIndoor:      util.Int32ptr(u.HumidityIndoor),
 		WindSpeedMph:        u.WindSpeedMPH,
 		WindGustMph:         u.WindGustMPH,
 		MaxDailyGust:        u.MaxDailyGust,
-		WindDir:             int32ptr(u.WindDir),
-		WindDirAvg_10M:      int32ptr(u.WindDirAvg10m),
-		Uv:                  int32ptr(u.UV),
+		WindDir:             util.Int32ptr(u.WindDir),
+		WindDirAvg_10M:      util.Int32ptr(u.WindDirAvg10m),
+		Uv:                  util.Int32ptr(u.UV),
 		SolarRadiation:      u.SolarRadiation,
 		HourlyRainIn:        u.HourlyRainIn,
 		EventRainIn:         u.EventRainIn,
@@ -52,7 +51,7 @@ func batteriesToPbBatteries(batteries []weather.BatteryStatus) []*pb.BatteryStat
 	for i, b := range batteries {
 		pbBatteries[i] = &pb.BatteryStatus{
 			Component: b.Component,
-			Status:    int32ptr(b.Status),
+			Status:    util.Int32ptr(b.Status),
 		}
 	}
 	return pbBatteries
@@ -64,22 +63,8 @@ func thSensorsToPbSensors(sensors []*weather.TempHumiditySensor) []*pb.TempHumid
 		pbSensors[i] = &pb.TempHumiditySensor{
 			Name:     s.Name,
 			TempF:    s.TempF,
-			Humidity: int32ptr(s.Humidity),
+			Humidity: util.Int32ptr(s.Humidity),
 		}
 	}
 	return pbSensors
 }
-
-func derefStr(s *string) string {
-	if s == nil {
-		return ""
-	}
-	return *s
-}
-
-func int32ptr(i *int) *int32 {
-	if i == nil {
-		return nil
-	}
-	return ptr.To(int32(*i))
-}

pkg/weather/grpc/mapupdate_test.go

@@ -0,0 +1,113 @@
+package grpc
+
+import (
+	"reflect"
+	"testing"
+	"time"
+
+	"k8s.io/utils/ptr"
+
+	pb "gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/api/v1alpha1/weather"
+	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/ambient/config"
+	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/weather"
+)
+
+var mockUpdate = &weather.WeatherUpdate{
+	DateUTC: &time.Time{},
+	StationConfig: &config.WeatherStation{
+		Name:      "50W",
+		Equipment: "WS-5000",
+	},
+	StationID:      ptr.To("50W"),
+	StationType:    ptr.To("WS-5000"),
+	TempOutdoorF:   ptr.To(97.6),
+	TempIndoorF:    ptr.To(77.6),
+	HumidityOudoor: ptr.To(50),
+	HumidityIndoor: ptr.To(50),
+	WindSpeedMPH:   ptr.To(20.5),
+	WindGustMPH:    ptr.To(30.5),
+	MaxDailyGust:   ptr.To(40.0),
+	WindDir:        ptr.To(180),
+	WindDirAvg10m:  ptr.To(180),
+	UV:             nil,
+	SolarRadiation: ptr.To(9.999),
+	HourlyRainIn:   ptr.To(9.999),
+	EventRainIn:    ptr.To(9.999),
+	DailyRainIn:    ptr.To(9.999),
+	WeeklyRainIn:   ptr.To(9.999),
+	MonthlyRainIn:  ptr.To(9.999),
+	YearlyRainIn:   ptr.To(9.999),
+	TotalRainIn:    ptr.To(9.999),
+	Batteries: []weather.BatteryStatus{
+		{Component: "battery1", Status: ptr.To(1)},
+		{Component: "battery2", Status: ptr.To(1)},
+	},
+	BaromRelativeIn: ptr.To(9.999),
+	BaromAbsoluteIn: ptr.To(9.999),
+	DewPointF:       ptr.To(9.999),
+	WindChillF:      ptr.To(9.999),
+	TempHumiditySensors: []*weather.TempHumiditySensor{
+		{Name: "sensor1", TempF: ptr.To(99.999), Humidity: nil},
+	},
+}
+
+func TestUpdateToPbUpdate(t *testing.T) {
+	type args struct {
+		u *weather.WeatherUpdate
+	}
+	tests := []struct {
+		name string
+		args args
+		want *pb.WeatherUpdate
+	}{
+		{
+			name: "Map Update to PB",
+			args: args{u: mockUpdate},
+			want: &pb.WeatherUpdate{
+				StationName:     mockUpdate.StationConfig.Name,
+				StationType:     *mockUpdate.StationType,
+				StationId:       *mockUpdate.StationID,
+				TempOutdoorF:    mockUpdate.TempOutdoorF,
+				TempIndoorF:     mockUpdate.TempIndoorF,
+				HumidityOutdoor: ptr.To(int32(*mockUpdate.HumidityOudoor)),
+				HumidityIndoor:  ptr.To(int32(*mockUpdate.HumidityIndoor)),
+				WindSpeedMph:    mockUpdate.WindSpeedMPH,
+				WindGustMph:     mockUpdate.WindGustMPH,
+				MaxDailyGust:    mockUpdate.MaxDailyGust,
+				WindDir:         ptr.To(int32(*mockUpdate.WindDir)),
+				WindDirAvg_10M:  ptr.To(int32(*mockUpdate.WindDirAvg10m)),
+				Uv:              nil,
+				SolarRadiation:  mockUpdate.SolarRadiation,
+				HourlyRainIn:    mockUpdate.HourlyRainIn,
+				EventRainIn:     mockUpdate.EventRainIn,
+				DailyRainIn:     mockUpdate.DailyRainIn,
+				WeeklyRainIn:    mockUpdate.WeeklyRainIn,
+				MonthlyRainIn:   mockUpdate.MonthlyRainIn,
+				YearlyRainIn:    mockUpdate.YearlyRainIn,
+				TotalRainIn:     mockUpdate.TotalRainIn,
+				Batteries: []*pb.BatteryStatus{
+					{Component: mockUpdate.Batteries[0].Component, Status: ptr.To(int32(*mockUpdate.Batteries[0].Status))},
+					{Component: mockUpdate.Batteries[1].Component, Status: ptr.To(int32(*mockUpdate.Batteries[1].Status))},
+				},
+				BaromRelativeIn: mockUpdate.BaromRelativeIn,
+				BaromAbsoluteIn: mockUpdate.BaromAbsoluteIn,
+				DewPointF:       mockUpdate.DewPointF,
+				WindChillF:      mockUpdate.WindChillF,
+				TempHumiditySensors: []*pb.TempHumiditySensor{
+					{
+						Name:     mockUpdate.TempHumiditySensors[0].Name,
+						TempF:    mockUpdate.TempHumiditySensors[0].TempF,
+						Humidity: nil,
+					},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := UpdateToPbUpdate(tt.args.u); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("UpdateToPbUpdate() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

pkg/weather/grpc/weather.go

@@ -13,21 +13,21 @@ import (
 	"gitea.libretechconsulting.com/rmcguire/go-app/pkg/otel"
 
 	pb "gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/api/v1alpha1/weather"
-	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/weather/state"
+	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/weather/recorder"
 )
 
 type GRPCWeather struct {
-	ctx    context.Context
-	state  *state.WeatherState
-	tracer trace.Tracer
+	ctx      context.Context
+	recorder *recorder.WeatherRecorder
+	tracer   trace.Tracer
 	*pb.UnimplementedAmbientLocalWeatherServiceServer
 }
 
-func NewGRPCWeather(ctx context.Context, state *state.WeatherState) *GRPCWeather {
+func NewGRPCWeather(ctx context.Context, recorder *recorder.WeatherRecorder) *GRPCWeather {
 	return &GRPCWeather{
-		ctx:    ctx,
-		state:  state,
-		tracer: otel.GetTracer(ctx, "grpcWeather"),
+		ctx:      ctx,
+		recorder: recorder,
+		tracer:   otel.GetTracer(ctx, "grpcWeather"),
 	}
 }
 
@@ -46,7 +46,7 @@ func (w *GRPCWeather) GetWeather(ctx context.Context, req *pb.GetWeatherRequest)
 
 	span.SetAttributes(attribute.Int("limit", limit))
 
-	updates, err := w.state.Get(ctx, limit)
+	updates, err := w.recorder.Get(ctx, req)
 	if err != nil {
 		span.RecordError(err)
 		span.SetStatus(otelcodes.Error, err.Error())

pkg/weather/recorder/recorder.go

@@ -0,0 +1,51 @@
+package recorder
+
+import (
+	"context"
+	"sync"
+
+	"go.opentelemetry.io/otel/metric"
+	"go.opentelemetry.io/otel/trace"
+
+	"gitea.libretechconsulting.com/rmcguire/go-app/pkg/otel"
+
+	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/weather/recorder/recorders"
+	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/weather/recorder/recorders/memory"
+)
+
+type WeatherRecorder struct {
+	recorder recorders.Recorder
+	ctx      context.Context
+	tracer   trace.Tracer
+	meter    metric.Meter
+	*sync.RWMutex
+}
+
+type Opts struct {
+	Ctx      context.Context
+	Recorder recorders.Recorder // If nil, will use memory recorder
+	KeepLast int
+}
+
+func NewWeatherRecorder(opts *Opts) *WeatherRecorder {
+	if opts.KeepLast < 1 {
+		opts.KeepLast = 1
+	}
+
+	if opts.Recorder == nil {
+		opts.Recorder = &memory.MemoryRecorder{}
+	}
+
+	opts.Recorder.Init(opts.Ctx, &recorders.RecorderOpts{
+		RetainLast: opts.KeepLast,
+		BaseCtx:    opts.Ctx,
+	})
+
+	return &WeatherRecorder{
+		ctx:      opts.Ctx,
+		recorder: opts.Recorder,
+		tracer:   otel.GetTracer(opts.Ctx, "weatherRecorder"),
+		meter:    otel.GetMeter(opts.Ctx, "weatherRecorder"),
+		RWMutex:  &sync.RWMutex{},
+	}
+}

pkg/weather/recorder/recorder_get.go

@@ -0,0 +1,65 @@
+package recorder
+
+import (
+	"context"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
+	"k8s.io/utils/ptr"
+
+	pb "gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/api/v1alpha1/weather"
+	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/weather"
+)
+
+// Returns last requested number of weather updates
+// If negative number given, will return all weather observations
+func (w *WeatherRecorder) Get(ctx context.Context, req *pb.GetWeatherRequest) (
+	[]*weather.WeatherUpdate, error,
+) {
+	if req == nil {
+		req = &pb.GetWeatherRequest{Limit: ptr.To(int32(-1))}
+	}
+
+	if req.Limit == nil || *req.Limit < 0 {
+		req.Limit = ptr.To(int32(-1))
+	} else if *req.Limit == 0 {
+		req.Limit = ptr.To(int32(1))
+	}
+
+	var filterSN, filterST string
+	if req.Opts != nil {
+		filterSN = req.Opts.GetStationName()
+		filterST = req.Opts.GetStationType()
+	}
+
+	ctx, span := w.tracer.Start(ctx, "getWeatherRecorder")
+	span.SetAttributes(
+		attribute.String("stationNameFilter", filterSN),
+		attribute.String("stationTypeFilter", filterST),
+		attribute.Int("last", int(*req.Limit)),
+		attribute.Int("currentSize", w.Count(ctx)),
+	)
+	defer span.End()
+
+	updates, err := w.recorder.Get(ctx, req)
+	if err != nil {
+		span.RecordError(err)
+		span.SetStatus(codes.Error, err.Error())
+	} else {
+		span.SetStatus(codes.Ok, "")
+	}
+
+	return updates, err
+}
+
+// Returns count of retained weather updates
+func (w *WeatherRecorder) Count(ctx context.Context) int {
+	ctx, span := w.tracer.Start(ctx, "countWeatherRecorder")
+	defer span.End()
+
+	count := w.recorder.Count(ctx)
+	span.SetAttributes(attribute.Int("count", count))
+
+	span.SetStatus(codes.Ok, "")
+	return count
+}

pkg/weather/recorder/recorder_set.go

@@ -0,0 +1,21 @@
+package recorder
+
+import (
+	"context"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
+
+	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/util"
+	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/weather"
+)
+
+func (w *WeatherRecorder) Set(ctx context.Context, u *weather.WeatherUpdate) error {
+	ctx, span := w.tracer.Start(ctx, "setRecorderUpdate", trace.WithAttributes(
+		attribute.String("stationName", u.StationConfig.Name),
+		attribute.String("stationType", util.DerefStr(u.StationType)),
+		attribute.String("stationEquipment", u.StationConfig.Equipment),
+	))
+	defer span.End()
+	return w.recorder.Set(ctx, u)
+}

pkg/weather/recorder/recorders/memory/get.go

@@ -0,0 +1,108 @@
+package memory
+
+import (
+	"context"
+	"errors"
+	"slices"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
+
+	pb "gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/api/v1alpha1/weather"
+	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/util"
+	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/weather"
+)
+
+func (r *MemoryRecorder) Get(ctx context.Context, req *pb.GetWeatherRequest) (
+	[]*weather.WeatherUpdate, error,
+) {
+	ctx, span := r.tracer.Start(ctx, "memoryRecorder.Get")
+	defer span.End()
+
+	r.RLock()
+	defer r.RUnlock()
+
+	span.AddEvent("acquired lock on recorder cache")
+
+	if r.count() == 0 {
+		err := errors.New("no recorded updates to get")
+		span.RecordError(err)
+		return nil, err
+	} else if r.count() <= int(*req.Limit) {
+		span.RecordError(errors.New("requested more updates than recorded"))
+	}
+
+	updates := r.getUpdatesFromReq(req)
+	span.AddEvent("request limit/opts applied to updates")
+
+	span.SetAttributes(attribute.Int("retrieved", len(updates)))
+	span.SetStatus(codes.Ok, "")
+
+	return updates, nil
+}
+
+func (r *MemoryRecorder) getUpdatesFromReq(req *pb.GetWeatherRequest) []*weather.WeatherUpdate {
+	if req.Opts == nil {
+		return limitUpdates(r.updates, int(req.GetLimit()))
+	}
+
+	return r.applyOptsToUpdates(r.updates, int(req.GetLimit()), req.Opts)
+}
+
+func (r *MemoryRecorder) applyOptsToUpdates(updates []*weather.WeatherUpdate, limit int, opts *pb.GetWeatherOpts) []*weather.WeatherUpdate {
+	if opts == nil {
+		return updates
+	} else if opts.StationName == nil && opts.StationType == nil {
+		return updates
+	}
+
+	filtered := make([]*weather.WeatherUpdate, 0, limit)
+
+	for i := len(updates) - 1; i >= 0; i-- {
+		update := updates[i]
+		match := true
+
+		if opts.GetStationName() != "" {
+			if update.GetStationName() != opts.GetStationName() {
+				match = false
+			}
+		}
+		if opts.GetStationType() != "" {
+			if util.DerefStr(update.StationType) != opts.GetStationType() {
+				match = false
+			}
+		}
+
+		if match {
+			filtered = append(filtered, update)
+			if len(filtered) >= limit {
+				return filtered
+			}
+		}
+	}
+
+	return slices.Clip(filtered)
+}
+
+func (r *MemoryRecorder) Count(ctx context.Context) int {
+	_, span := r.tracer.Start(ctx, "countWeatherRecorder")
+	defer span.End()
+
+	count := r.count()
+
+	span.SetAttributes(attribute.Int("count", count))
+	span.SetStatus(codes.Ok, "")
+
+	return count
+}
+
+func (r *MemoryRecorder) count() int {
+	return len(r.updates)
+}
+
+func limitUpdates(updates []*weather.WeatherUpdate, limit int) []*weather.WeatherUpdate {
+	if len(updates) > limit {
+		return updates[len(updates)-limit:]
+	}
+	return updates
+}

pkg/weather/recorder/recorders/memory/memory.go

@@ -0,0 +1,35 @@
+package memory
+
+import (
+	"context"
+	"sync"
+
+	"go.opentelemetry.io/otel/trace"
+
+	"gitea.libretechconsulting.com/rmcguire/go-app/pkg/otel"
+
+	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/weather"
+	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/weather/recorder/recorders"
+)
+
+const defRetainLast = 120
+
+type MemoryRecorder struct {
+	baseCtx context.Context
+	updates []*weather.WeatherUpdate
+	tracer  trace.Tracer
+	keep    int
+	*sync.RWMutex
+}
+
+func (r *MemoryRecorder) Init(ctx context.Context, opts *recorders.RecorderOpts) {
+	if opts.RetainLast < 1 {
+		opts.RetainLast = defRetainLast
+	}
+
+	r.updates = make([]*weather.WeatherUpdate, 0, opts.RetainLast)
+	r.keep = opts.RetainLast
+	r.baseCtx = opts.BaseCtx
+	r.RWMutex = &sync.RWMutex{}
+	r.tracer = otel.GetTracer(r.baseCtx, "memoryRecorder")
+}

pkg/weather/recorder/recorders/memory/set.go

@@ -0,0 +1,38 @@
+package memory
+
+import (
+	"context"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
+
+	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/weather"
+)
+
+func (r *MemoryRecorder) Set(ctx context.Context, u *weather.WeatherUpdate) error {
+	r.Lock()
+	defer r.Unlock()
+
+	ctx, span := r.tracer.Start(ctx, "memoryRecorder.Set")
+	span.SetAttributes(
+		attribute.Int("countWeatherUpdates", r.Count(ctx)),
+		attribute.Int("keepUpdates", r.keep),
+	)
+	defer span.End()
+
+	return r.set(ctx, u)
+}
+
+func (r *MemoryRecorder) set(ctx context.Context, u *weather.WeatherUpdate) error {
+	span := trace.SpanFromContext(ctx)
+
+	if len(r.updates) > r.keep {
+		r.updates = r.updates[1:]
+		span.AddEvent("trimmed recorded updates by 1")
+	}
+
+	r.updates = append(r.updates, u)
+	span.AddEvent("recorded weather update")
+
+	return nil
+}

pkg/weather/recorder/recorders/recorders.go

@@ -0,0 +1,20 @@
+package recorders
+
+import (
+	"context"
+
+	pb "gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/api/v1alpha1/weather"
+	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/weather"
+)
+
+type RecorderOpts struct {
+	RetainLast int
+	BaseCtx    context.Context
+}
+
+type Recorder interface {
+	Init(context.Context, *RecorderOpts)
+	Set(context.Context, *weather.WeatherUpdate) error
+	Get(context.Context, *pb.GetWeatherRequest) ([]*weather.WeatherUpdate, error)
+	Count(context.Context) int // Best Effort
+}

pkg/weather/state/state.go

@@ -1,41 +0,0 @@
-package state
-
-import (
-	"context"
-	"sync"
-
-	"go.opentelemetry.io/otel/metric"
-	"go.opentelemetry.io/otel/trace"
-
-	"gitea.libretechconsulting.com/rmcguire/go-app/pkg/otel"
-
-	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/weather"
-)
-
-type WeatherState struct {
-	updates []*weather.WeatherUpdate
-	keep    int
-	ctx     context.Context
-	tracer  trace.Tracer
-	meter   metric.Meter
-	*sync.RWMutex
-}
-
-type Opts struct {
-	Ctx      context.Context
-	KeepLast int
-}
-
-func NewWeatherState(opts *Opts) *WeatherState {
-	if opts.KeepLast < 1 {
-		opts.KeepLast = 1
-	}
-
-	return &WeatherState{
-		updates: make([]*weather.WeatherUpdate, 0),
-		keep:    opts.KeepLast,
-		ctx:     opts.Ctx,
-		tracer:  otel.GetTracer(opts.Ctx, "weatherState"),
-		meter:   otel.GetMeter(opts.Ctx, "weatherState"),
-	}
-}

pkg/weather/state/state_get.go

@@ -1,87 +0,0 @@
-package state
-
-import (
-	"context"
-	"errors"
-
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/codes"
-	"go.opentelemetry.io/otel/trace"
-
-	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/weather"
-)
-
-// Returns last requested number of weather updates
-func (w *WeatherState) Get(ctx context.Context, last int) (
-	[]*weather.WeatherUpdate, error,
-) {
-	if last < 1 {
-		last = 1
-	}
-
-	ctx, span := w.tracer.Start(ctx, "getWeatherState")
-	span.SetAttributes(
-		attribute.Int("last", last),
-		attribute.Int("keep", w.keep),
-		attribute.Int("currentSize", w.Count()),
-	)
-	defer span.End()
-
-	updates, err := w.get(ctx, last)
-	if err != nil {
-		span.RecordError(err)
-		span.SetStatus(codes.Error, err.Error())
-	} else {
-		span.SetStatus(codes.Ok, "")
-	}
-
-	return updates, err
-}
-
-func (w *WeatherState) get(ctx context.Context, last int) (
-	[]*weather.WeatherUpdate, error,
-) {
-	span := trace.SpanFromContext(ctx)
-
-	w.RLock()
-	defer w.Unlock()
-
-	span.AddEvent("acquired lock on state cache")
-
-	updates := w.updates
-
-	if w.count() == 0 {
-		err := errors.New("no state to get")
-		span.RecordError(err)
-		return nil, err
-	} else if w.count() <= last {
-		span.RecordError(errors.New("requested more state than exists"))
-	} else {
-		updates = updates[len(updates)-last:]
-	}
-
-	span.SetAttributes(attribute.Int("retrieved", len(updates)))
-	span.SetStatus(codes.Ok, "")
-
-	return updates, nil
-}
-
-// Returns count of retained weather updates
-func (w *WeatherState) Count() int {
-	_, span := w.tracer.Start(w.ctx, "countWeatherState")
-	defer span.End()
-
-	count := w.count()
-
-	span.SetAttributes(attribute.Int("count", count))
-	span.SetStatus(codes.Ok, "")
-
-	return count
-}
-
-func (w *WeatherState) count() int {
-	w.RLock()
-	defer w.RUnlock()
-
-	return len(w.updates)
-}

pkg/weather/state/state_set.go

@@ -1,35 +0,0 @@
-package state
-
-import (
-	"context"
-
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/trace"
-
-	"gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/weather"
-)
-
-func (w *WeatherState) Set(ctx context.Context, u *weather.WeatherUpdate) error {
-	_, span := w.tracer.Start(ctx, "setWeatherState")
-	span.SetAttributes(
-		attribute.Int("countWeatherUpdates", w.Count()),
-		attribute.Int("keepUpdates", w.keep),
-	)
-	defer span.End()
-
-	return w.set(span, u)
-}
-
-func (w *WeatherState) set(span trace.Span, u *weather.WeatherUpdate) error {
-	w.Lock()
-	defer w.Unlock()
-
-	if len(w.updates) > w.keep {
-		w.updates = w.updates[1:]
-		span.AddEvent("trimmed state updates by 1")
-	}
-
-	w.updates = append(w.updates, u)
-	span.AddEvent("recorded weather state")
-	return nil
-}