Compare commits

...

3 Commits

Author SHA1 Message Date
0b27285b86 update gitea ci
Some checks failed
Build and Publish / release (push) Has been skipped
Build and Publish / chart-updated (push) Failing after 18s
Build and Publish / helm-release (push) Has been skipped
2025-03-21 13:08:08 -04:00
1f097b1fd7 fix recorder 2025-03-21 12:52:39 -04:00
42eea2346b refactor state to recorder, finish implementing 2025-03-21 09:22:32 -04:00
8 changed files with 116 additions and 64 deletions

View File

@ -25,7 +25,7 @@ jobs:
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
@ -89,11 +89,29 @@ jobs:
VER_PKG=${{ env.VER_PKG }}
VERSION=${{ github.ref_name }}
# Detect if the helm chart was updated
chart-updated:
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:
- **/Chart.yaml
helm-release:
runs-on: ubuntu-latest
needs: chart-updated
if: ${{ needs.chart-updated.outputs.chart-updated == 'true' }}
steps:
- name: Checkout Code
uses: actions/checkout@v3
uses: actions/checkout@v4
- name: Install Helm
env:
@ -103,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:

View File

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

20
main.go
View File

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

View File

@ -20,35 +20,63 @@ 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 {
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))
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 +106,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 +141,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 +276,10 @@ func (aw *AmbientWeather) enrichStation(update *weather.WeatherUpdate) {
}
}
}
func (aw *AmbientWeather) GetState() *recorder.WeatherRecorder {
aw.RLock()
defer aw.RUnlock()
return aw.weatherState
}

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

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

View File

@ -1,4 +1,4 @@
package state
package recorder
import (
"context"
@ -12,18 +12,21 @@ 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),
attribute.Int("currentSize", w.Count()),
attribute.Int("currentSize", w.Count(ctx)),
)
defer span.End()
@ -38,24 +41,24 @@ 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)
w.RLock()
defer w.Unlock()
defer w.RUnlock()
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(ctx context.Context) int {
_, span := w.tracer.Start(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()

View File

@ -1,4 +1,4 @@
package state
package recorder
import (
"context"
@ -9,10 +9,10 @@ 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 {
ctx, span := w.tracer.Start(ctx, "recordWeatherUpdate")
span.SetAttributes(
attribute.Int("countWeatherUpdates", w.Count()),
attribute.Int("countWeatherUpdates", w.Count(ctx)),
attribute.Int("keepUpdates", w.keep),
)
defer span.End()
@ -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
}