gsn/predictor
1
Fork 1
Code Issues 5 Pull requests 1 Projects Releases Activity

feat: remove redis

Browse source
This commit is contained in:
Anatoly Antonov 2025-10-20 16:31:45 +09:00
parent 7a9f81e527
commit a850615e1f
18 changed files with 170 additions and 1142 deletions
Download patch file Download diff file Expand all files Collapse all files

4
internal/pkg/errcodes/errcodes.go
View file

@ -91,10 +91,6 @@ var (
ErrConfig = New(http.StatusInternalServerError, "configuration error")
ErrConfigInvalidEnv = New(http.StatusInternalServerError, "invalid environment configuration")
ErrConfigMissingRequired = New(http.StatusInternalServerError, "missing required configuration")
ErrRedis = New(http.StatusInternalServerError, "redis error")
ErrRedisLockAlreadyLocked = New(http.StatusConflict, "could not perform redis lock", "already locked")
ErrRedisCacheMiss = New(http.StatusNotFound, "cache miss", "key not found")
ErrRedisCacheCorrupted = New(http.StatusInternalServerError, "cache data corrupted", "invalid format")
ErrDownload = New(http.StatusInternalServerError, "download error")
ErrProcessing = New(http.StatusInternalServerError, "data processing error")
ErrNoCubeFilesFound = New(http.StatusNotFound, "no cube files found")

8
internal/pkg/grib/README.md
View file

@ -8,7 +8,7 @@
- **Сборка 5D-куба** (время, давление, широта, долгота, переменные u/v)
- **Эффективное хранение** с использованием mmap
- **Интерполяция** ветровых данных для произвольных координат и времени
- **Кэширование** результатов (in-memory + Redis)
- **Кэширование** результатов (in-memory)
- **Распределенные блокировки** для предотвращения дублирования загрузок
## Архитектура
@ -38,7 +38,6 @@ cfg := grib.ServiceConfig{
Dir: "/tmp/grib",
TTL: 24 * time.Hour,
CacheTTL: 1 * time.Hour,
Redis: redisClient,
Parallel: 4,
Client: &http.Client{Timeout: 30 * time.Second},
}
@ -68,7 +67,6 @@ wind, err := service.Extract(ctx, lat, lon, alt, timestamp)
## Кэширование
- **In-memory кэш**: быстрый доступ к недавно запрошенным данным
- **Redis кэш**: распределенное кэширование для множественных реплик
## Расписание обновлений
@ -83,12 +81,11 @@ wind, err := service.Extract(ctx, lat, lon, alt, timestamp)
- **Высокая производительность** (mmap, конкурентные загрузки)
- **Эффективное использование памяти** (не загружает весь массив в RAM)
- **Горизонтальное масштабирование** (stateless, множество реплик)
- **Встроенное кэширование** (in-memory + Redis)
- **Встроенное кэширование** (in-memory)
### Особенности:
- Использует `github.com/nilsmagnus/grib` вместо pygrib
- Реализует собственную логику интерполяции
- Поддерживает распределенные блокировки через Redis
## Конфигурация
@ -99,6 +96,5 @@ wind, err := service.Extract(ctx, lat, lon, alt, timestamp)
- `Dir` - директория для хранения файлов
- `TTL` - время жизни данных (по умолчанию 24 часа)
- `CacheTTL` - время жизни кэша (по умолчанию 1 час)
- `Redis` - Redis клиент для блокировок и кэша
- `Parallel` - количество параллельных загрузок
- `Client` - HTTP клиент для загрузок

43
internal/pkg/grib/grib.go
View file

@ -3,7 +3,6 @@ package grib
import (
"context"
"encoding/binary"
"fmt"
"math"
"net/http"
"os"
@ -17,12 +16,6 @@ import (
"github.com/nilsmagnus/grib/griblib"
)
type RedisIface interface {
Lock(ctx context.Context, key string, ttl time.Duration) (func(context.Context), error)
Set(key string, value []byte, ttl time.Duration) error
Get(key string) ([]byte, error)
}
type Service interface {
Update(ctx context.Context) error
Extract(ctx context.Context, lat, lon, alt float64, ts time.Time) ([2]float64, error)
@ -34,7 +27,6 @@ type ServiceConfig struct {
Dir string
TTL time.Duration
CacheTTL time.Duration
Redis RedisIface
Parallel int
Client *http.Client
DatasetURL string
@ -134,12 +126,6 @@ func (s *service) Update(ctx context.Context) error {
}
}
unlock, err := s.cfg.Redis.Lock(ctx, "grib-dl", 45*time.Minute)
if err != nil {
return err
}
defer unlock(ctx)
// Check again after acquiring lock (double-checked locking pattern)
if d := s.data.Load(); d != nil {
runTime := time.Unix(d.runUTC, 0)
@ -290,29 +276,6 @@ func (s *service) Extract(ctx context.Context, lat, lon, alt float64, ts time.Ti
return [2]float64(v), nil
}
// Try Redis cache
redisKey := fmt.Sprintf("grib:extract:%d", key)
if cached, err := s.cfg.Redis.Get(redisKey); err == nil {
var result [2]float64
if len(cached) == 16 {
result[0] = math.Float64frombits(binary.LittleEndian.Uint64(cached[:8]))
result[1] = math.Float64frombits(binary.LittleEndian.Uint64(cached[8:]))
s.cache.set(key, vec(result))
return result, nil
} else {
// Cache data is corrupted (wrong length)
return zero, errcodes.ErrRedisCacheCorrupted
}
} else {
// Check if it's a cache miss (expected error)
if errcode, ok := errcodes.AsErr(err); ok && errcode == errcodes.ErrRedisCacheMiss {
// Cache miss is expected, continue with calculation
} else {
// Unexpected error, return it
return zero, err
}
}
// Calculate result
td := ts.Sub(time.Unix(d.runUTC, 0)).Hours()
u, v := d.uv(lat, lon, alt, td)
@ -321,12 +284,6 @@ func (s *service) Extract(ctx context.Context, lat, lon, alt float64, ts time.Ti
// Cache in memory
s.cache.set(key, vec(out))
// Cache in Redis
encoded := make([]byte, 16)
binary.LittleEndian.PutUint64(encoded[:8], math.Float64bits(out[0]))
binary.LittleEndian.PutUint64(encoded[8:], math.Float64bits(out[1]))
s.cfg.Redis.Set(redisKey, encoded, s.cfg.CacheTTL)
return out, nil
}

6
internal/service/config.go
View file

@ -13,12 +13,6 @@ type Config struct {
GribParallel int `env:"GSN_PREDICTOR_GRIB_PARALLEL" envDefault:"4"`
GribTimeout time.Duration `env:"GSN_PREDICTOR_GRIB_TIMEOUT" envDefault:"30s"`
GribDatasetURL string `env:"GSN_PREDICTOR_GRIB_DATASET_URL" envDefault:"https://nomads.ncep.noaa.gov/pub/data/nccf/com/gfs/prod"`
// --- Redis Configuration ---
RedisHost string `env:"GSN_PREDICTOR_REDIS_HOST"`
RedisPort int `env:"GSN_PREDICTOR_REDIS_PORT"`
RedisPassword string `env:"GSN_PREDICTOR_REDIS_PASSWORD"`
RedisDB int `env:"GSN_PREDICTOR_REDIS_DB"`
}
func (c *Config) CreateHTTPClient() *http.Client {

7
internal/service/deps.go
View file

@ -10,10 +10,3 @@ type Grib interface {
Extract(ctx context.Context, lat, lon, alt float64, ts time.Time) ([2]float64, error)
Close() error
}
type Redis interface {
Lock(ctx context.Context, key string, ttl time.Duration) (func(context.Context), error)
Set(key string, value []byte, ttl time.Duration) error
Get(key string) ([]byte, error)
Close() error
}

85
internal/service/predictor.go
View file

@ -219,6 +219,31 @@ func (s *Service) customProfile(ctx context.Context, params ds.PredictionParamet
return results
}
func rk4Step(lat, lon, alt float64, t time.Time, dt float64, windFunc func(lat, lon, alt float64, t time.Time) (float64, float64), altRate float64) (float64, float64, float64) {
// Helper for RK4 integration step
toRad := math.Pi / 180.0
toDeg := 180.0 / math.Pi
R := func(alt float64) float64 { return 6371009.0 + alt }
f := func(lat, lon, alt float64, t time.Time) (float64, float64, float64) {
windU, windV := windFunc(lat, lon, alt, t)
Rnow := R(alt)
dlat := toDeg * windV / Rnow
dlon := toDeg * windU / (Rnow * math.Cos(lat*toRad))
return dlat, dlon, altRate
}
k1_lat, k1_lon, k1_alt := f(lat, lon, alt, t)
k2_lat, k2_lon, k2_alt := f(lat+0.5*k1_lat*dt, lon+0.5*k1_lon*dt, alt+0.5*k1_alt*dt, t.Add(time.Duration(0.5*dt)*time.Second))
k3_lat, k3_lon, k3_alt := f(lat+0.5*k2_lat*dt, lon+0.5*k2_lon*dt, alt+0.5*k2_alt*dt, t.Add(time.Duration(0.5*dt)*time.Second))
k4_lat, k4_lon, k4_alt := f(lat+k3_lat*dt, lon+k3_lon*dt, alt+k3_alt*dt, t.Add(time.Duration(dt)*time.Second))
latNew := lat + (dt/6.0)*(k1_lat+2*k2_lat+2*k3_lat+k4_lat)
lonNew := lon + (dt/6.0)*(k1_lon+2*k2_lon+2*k3_lon+k4_lon)
altNew := alt + (dt/6.0)*(k1_alt+2*k2_alt+2*k3_alt+k4_alt)
return latNew, lonNew, altNew
}
func (s *Service) simulateAscent(ctx context.Context, params ds.PredictionParameters, ascentRate, targetAltitude float64, customCurve *CustomCurve) []ds.PredicitonResult {
const dt = 10.0 // simulation step in seconds
const outputInterval = 60.0 // output every 60 seconds
@ -230,7 +255,6 @@ func (s *Service) simulateAscent(ctx context.Context, params ds.PredictionParame
results := make([]ds.PredicitonResult, 0, 1000)
// Always include the initial launch point
latCopy := lat
lonCopy := lon
altCopy := alt
@ -247,40 +271,39 @@ func (s *Service) simulateAscent(ctx context.Context, params ds.PredictionParame
WindV: &windV,
})
var nextOutputTime = timeCur.Add(time.Duration(outputInterval) * time.Second)
for alt < targetAltitude {
wind, err := s.ExtractWind(ctx, lat, lon, alt, timeCur)
nextOutputTime := timeCur.Add(time.Duration(outputInterval) * time.Second)
windFunc := func(lat, lon, alt float64, t time.Time) (float64, float64) {
w, err := s.ExtractWind(ctx, lat, lon, alt, t)
if err != nil {
log.Ctx(ctx).Warn("Wind extraction failed during ascent", zap.Error(err))
break
return 0, 0
}
return w[0], w[1]
}
for alt < targetAltitude {
altRate := ascentRate
if customCurve != nil {
altRate = s.getCustomAltitudeRate(customCurve, alt, ascentRate)
}
latDot := (wind[1] / 111320.0)
lonDot := (wind[0] / (40075000.0 * math.Cos(lat*math.Pi/180) / 360.0))
lat += latDot * dt
lon += lonDot * dt
alt += altRate * dt
latNew, lonNew, altNew := rk4Step(lat, lon, alt, timeCur, dt, windFunc, altRate)
timeCur = timeCur.Add(time.Duration(dt) * time.Second)
lat = latNew
lon = lonNew
alt = altNew
// Don't add a point if we've reached or exceeded target altitude
if alt >= targetAltitude {
break
}
if !timeCur.Before(nextOutputTime) {
wU, wV := windFunc(lat, lon, alt, timeCur)
latCopy := lat
lonCopy := lon
altCopy := alt
timeCopy := timeCur
windU := wind[0]
windV := wind[1]
windU := wU
windV := wV
results = append(results, ds.PredicitonResult{
Latitude: &latCopy,
Longitude: &lonCopy,
@ -307,7 +330,6 @@ func (s *Service) simulateDescent(ctx context.Context, params ds.PredictionParam
results := make([]ds.PredicitonResult, 0, 1000)
// Always include the initial descent point
latCopy := lat
lonCopy := lon
altCopy := alt
@ -324,40 +346,39 @@ func (s *Service) simulateDescent(ctx context.Context, params ds.PredictionParam
WindV: &windV,
})
var nextOutputTime = timeCur.Add(time.Duration(outputInterval) * time.Second)
for alt > targetAltitude {
wind, err := s.ExtractWind(ctx, lat, lon, alt, timeCur)
nextOutputTime := timeCur.Add(time.Duration(outputInterval) * time.Second)
windFunc := func(lat, lon, alt float64, t time.Time) (float64, float64) {
w, err := s.ExtractWind(ctx, lat, lon, alt, t)
if err != nil {
log.Ctx(ctx).Warn("Wind extraction failed during descent", zap.Error(err))
break
return 0, 0
}
return w[0], w[1]
}
for alt > targetAltitude {
altRate := -descentRate
if customCurve != nil {
altRate = -s.getCustomAltitudeRate(customCurve, alt, descentRate)
}
latDot := (wind[1] / 111320.0)
lonDot := (wind[0] / (40075000.0 * math.Cos(lat*math.Pi/180) / 360.0))
lat += latDot * dt
lon += lonDot * dt
alt += altRate * dt
latNew, lonNew, altNew := rk4Step(lat, lon, alt, timeCur, dt, windFunc, altRate)
timeCur = timeCur.Add(time.Duration(dt) * time.Second)
lat = latNew
lon = lonNew
alt = altNew
// Don't add a point if we've reached or gone below target altitude
if alt <= targetAltitude {
break
}
if !timeCur.Before(nextOutputTime) {
wU, wV := windFunc(lat, lon, alt, timeCur)
latCopy := lat
lonCopy := lon
altCopy := alt
timeCopy := timeCur
windU := wind[0]
windV := wind[1]
windU := wU
windV := wV
results = append(results, ds.PredicitonResult{
Latitude: &latCopy,
Longitude: &lonCopy,

12
internal/service/service.go
View file

@ -8,16 +8,14 @@ import (
)
type Service struct {
cfg *Config
redis Redis
grib Grib
cfg *Config
grib Grib
}
func New(cfg *Config, gribService Grib, redisService Redis) (*Service, error) {
func New(cfg *Config, gribService Grib) (*Service, error) {
svc := &Service{
cfg: cfg,
redis: redisService,
grib: gribService,
cfg: cfg,
grib: gribService,
}
return svc, nil