From 42eea2346bd05e2b7ef6ac401511beb924448fa1 Mon Sep 17 00:00:00 2001 From: Ryan D McGuire Date: Fri, 21 Mar 2025 09:22:32 -0400 Subject: [PATCH] refactor state to recorder, finish implementing --- docker-compose-sample.yml | 3 +- main.go | 20 ++----- pkg/ambient/ambient.go | 52 ++++++++++++++++--- pkg/weather/grpc/weather.go | 18 +++---- .../{state/state.go => recorder/recorder.go} | 15 +++--- .../state_get.go => recorder/recorder_get.go} | 25 +++++---- .../state_set.go => recorder/recorder_set.go} | 12 ++--- 7 files changed, 89 insertions(+), 56 deletions(-) rename pkg/weather/{state/state.go => recorder/recorder.go} (61%) rename pkg/weather/{state/state_get.go => recorder/recorder_get.go} (67%) rename pkg/weather/{state/state_set.go => recorder/recorder_set.go} (59%) diff --git a/docker-compose-sample.yml b/docker-compose-sample.yml index 39b6d77..3f74032 100644 --- a/docker-compose-sample.yml +++ b/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: diff --git a/main.go b/main.go index 84e5d60..ceb8212 100644 --- a/main.go +++ b/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.GetState()), }, }, }, diff --git a/pkg/ambient/ambient.go b/pkg/ambient/ambient.go index c4ddccb..195459a 100644 --- a/pkg/ambient/ambient.go +++ b/pkg/ambient/ambient.go @@ -20,19 +20,24 @@ 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 + weatherState *recorder.WeatherRecorder + appCtx context.Context + metrics *weather.WeatherMetrics + l *zerolog.Logger + *sync.RWMutex } func New(appCtx context.Context, awConfig *config.AmbientLocalExporterConfig) *AmbientWeather { @@ -44,11 +49,33 @@ func New(appCtx context.Context, awConfig *config.AmbientLocalExporterConfig) *A // 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)) + + aw.weatherState = 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 +105,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 +140,9 @@ func (aw *AmbientWeather) handleProviderRequest( updateSpan.SetAttributes(attribute.String("stationName", update.StationConfig.Name)) } + // Record state + aw.weatherState.Set(ctx, update) + // Update metrics aw.metricsUpdate(ctx, p, update) @@ -242,3 +275,10 @@ func (aw *AmbientWeather) enrichStation(update *weather.WeatherUpdate) { } } } + +func (aw *AmbientWeather) GetState() *recorder.WeatherRecorder { + aw.RLock() + defer aw.RUnlock() + + return aw.weatherState +} diff --git a/pkg/weather/grpc/weather.go b/pkg/weather/grpc/weather.go index c41928d..28c0bf6 100644 --- a/pkg/weather/grpc/weather.go +++ b/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, limit) if err != nil { span.RecordError(err) span.SetStatus(otelcodes.Error, err.Error()) diff --git a/pkg/weather/state/state.go b/pkg/weather/recorder/recorder.go similarity index 61% rename from pkg/weather/state/state.go rename to pkg/weather/recorder/recorder.go index 1148b6b..9b29d87 100644 --- a/pkg/weather/state/state.go +++ b/pkg/weather/recorder/recorder.go @@ -1,4 +1,4 @@ -package state +package recorder import ( "context" @@ -12,7 +12,7 @@ import ( "gitea.libretechconsulting.com/rmcguire/ambient-local-exporter/pkg/weather" ) -type WeatherState struct { +type WeatherRecorder struct { updates []*weather.WeatherUpdate keep int ctx context.Context @@ -26,16 +26,17 @@ type Opts struct { KeepLast int } -func NewWeatherState(opts *Opts) *WeatherState { +func NewWeatherRecorder(opts *Opts) *WeatherRecorder { if opts.KeepLast < 1 { opts.KeepLast = 1 } - return &WeatherState{ - updates: make([]*weather.WeatherUpdate, 0), + return &WeatherRecorder{ + updates: make([]*weather.WeatherUpdate, 0, opts.KeepLast), keep: opts.KeepLast, ctx: opts.Ctx, - tracer: otel.GetTracer(opts.Ctx, "weatherState"), - meter: otel.GetMeter(opts.Ctx, "weatherState"), + tracer: otel.GetTracer(opts.Ctx, "weatherRecorder"), + meter: otel.GetMeter(opts.Ctx, "weatherRecorder"), + RWMutex: &sync.RWMutex{}, } } diff --git a/pkg/weather/state/state_get.go b/pkg/weather/recorder/recorder_get.go similarity index 67% rename from pkg/weather/state/state_get.go rename to pkg/weather/recorder/recorder_get.go index c2d5fcd..9839f04 100644 --- a/pkg/weather/state/state_get.go +++ b/pkg/weather/recorder/recorder_get.go @@ -1,4 +1,4 @@ -package state +package recorder import ( "context" @@ -12,14 +12,17 @@ import ( ) // Returns last requested number of weather updates -func (w *WeatherState) Get(ctx context.Context, last int) ( +// If negative number given, will return all weather observations +func (w *WeatherRecorder) Get(ctx context.Context, last int) ( []*weather.WeatherUpdate, error, ) { - if last < 1 { + if last < 0 { + last = w.keep + } else if last < 1 { last = 1 } - ctx, span := w.tracer.Start(ctx, "getWeatherState") + ctx, span := w.tracer.Start(ctx, "getWeatherRecorder") span.SetAttributes( attribute.Int("last", last), attribute.Int("keep", w.keep), @@ -38,7 +41,7 @@ func (w *WeatherState) Get(ctx context.Context, last int) ( return updates, err } -func (w *WeatherState) get(ctx context.Context, last int) ( +func (w *WeatherRecorder) get(ctx context.Context, last int) ( []*weather.WeatherUpdate, error, ) { span := trace.SpanFromContext(ctx) @@ -46,16 +49,16 @@ func (w *WeatherState) get(ctx context.Context, last int) ( w.RLock() defer w.Unlock() - span.AddEvent("acquired lock on state cache") + span.AddEvent("acquired lock on recorder cache") updates := w.updates if w.count() == 0 { - err := errors.New("no state to get") + err := errors.New("no recorded updates to get") span.RecordError(err) return nil, err } else if w.count() <= last { - span.RecordError(errors.New("requested more state than exists")) + span.RecordError(errors.New("requested more updates than recorded")) } else { updates = updates[len(updates)-last:] } @@ -67,8 +70,8 @@ func (w *WeatherState) get(ctx context.Context, last int) ( } // Returns count of retained weather updates -func (w *WeatherState) Count() int { - _, span := w.tracer.Start(w.ctx, "countWeatherState") +func (w *WeatherRecorder) Count() int { + _, span := w.tracer.Start(w.ctx, "countWeatherRecorder") defer span.End() count := w.count() @@ -79,7 +82,7 @@ func (w *WeatherState) Count() int { return count } -func (w *WeatherState) count() int { +func (w *WeatherRecorder) count() int { w.RLock() defer w.RUnlock() diff --git a/pkg/weather/state/state_set.go b/pkg/weather/recorder/recorder_set.go similarity index 59% rename from pkg/weather/state/state_set.go rename to pkg/weather/recorder/recorder_set.go index 2c4a5d6..d5a634a 100644 --- a/pkg/weather/state/state_set.go +++ b/pkg/weather/recorder/recorder_set.go @@ -1,4 +1,4 @@ -package state +package recorder import ( "context" @@ -9,8 +9,8 @@ import ( "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") +func (w *WeatherRecorder) Set(ctx context.Context, u *weather.WeatherUpdate) error { + _, span := w.tracer.Start(ctx, "recordWeatherUpdate") span.SetAttributes( attribute.Int("countWeatherUpdates", w.Count()), attribute.Int("keepUpdates", w.keep), @@ -20,16 +20,16 @@ func (w *WeatherState) Set(ctx context.Context, u *weather.WeatherUpdate) error return w.set(span, u) } -func (w *WeatherState) set(span trace.Span, u *weather.WeatherUpdate) error { +func (w *WeatherRecorder) 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") + span.AddEvent("trimmed recorded updates by 1") } w.updates = append(w.updates, u) - span.AddEvent("recorded weather state") + span.AddEvent("recorded weather update") return nil }