rmcguire/ambient-local-exporter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor state to recorder, finish implementing

Browse Source
This commit is contained in:
Ryan McGuire
2025-03-21 09:22:32 -04:00
parent a5948cf334
commit 42eea2346b
7 changed files with 89 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
pkg/ambient/ambient.go
View File

@ -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
}

18
pkg/weather/grpc/weather.go
View File

@ -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())

15
pkg/weather/state/state.go → pkg/weather/recorder/recorder.go
View File

@ -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{},
}
}

25
pkg/weather/state/state_get.go → pkg/weather/recorder/recorder_get.go
View File

@ -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()

12
pkg/weather/state/state_set.go → pkg/weather/recorder/recorder_set.go
View File

@ -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
}