Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions correctedforecaster/internal/server/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -148,6 +148,8 @@ func (s *Server) correctWithBetterTopography(request *internalprotocol.GetForeca
}

altitudeDiff := *modelAltitude - realAltitude
// Only correct when the altitude difference exceeds ±100 m. Smaller gaps are
// within the noise of the model grid and not worth adjusting.
if altitudeDiff < -100 || altitudeDiff > 100 {
correction.UpdateTemperature(interpreted, altitudeDiff)
correction.UpdateDewpointTemperature(interpreted)
Expand Down
4 changes: 4 additions & 0 deletions healthz/internal/health/probe.go
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,10 @@ func NewProbeResult(serviceProblems, dataProblems map[string][]string, maxfailed
func runProbe(conf *config.ProbeConfiguration) ProbeResult {
log.Println("Perform probe...")

// serviceProblems and dataProblems are tracked separately so that health can be
// reported on two independent dimensions: service (any location failing means the
// API itself is broken) vs. data (a configurable number of locations may have stale
// or missing data before the data dimension is considered unhealthy).
serviceProblems := map[string][]string{}
dataProblems := map[string][]string{}

Expand Down
6 changes: 5 additions & 1 deletion jsonfrontend/internal/server/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,9 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
metrics.OutsideAllGrids.Add(1)
return
case internalprotocol.ForecastStatus_PointTooFarAway:
// The point is within a grid's bounding area but too far from any grid cell.
// Return 200 with a GeoJSON error body rather than 4xx so that clients receive
// a structured response with location context instead of a plain HTTP error.
doc = encode.EncodeError(data, "no data at the given location")
metrics.PointTooFarAway.Add(1)
default:
Expand Down Expand Up @@ -193,7 +196,8 @@ func getParam(q url.Values, name string, from float32, to float32) (float32, err
}

// expires returns a randomized time.Time for use as a value to the Expires header.
// Maximum value is current + maxOffset seconds.
// The expiry is randomly drawn from [0.8×maxOffset, maxOffset) rather than being
// a fixed value. This spreads cache invalidation across clients to prevent stampedes.
// Returns current time + 1 minute if maxOffset <=0.
func expires(current time.Time, maxOffset int) time.Time {
if maxOffset > 0 {
Expand Down
57 changes: 11 additions & 46 deletions rawdataforecaster/README.md
Original file line number Diff line number Diff line change
@@ -1,53 +1,18 @@
# Simple Forecaster
# rawdataforecaster

Serves unmodified data from a forecast. The term 'simple' refers to the data not having been processed by forti.
Serves raw (uncorrected) forecast data over gRPC from a blob store (Azure, S3, or local `file://`) in the [forti-internalformat](https://github.com/metno/forti-internalformat) format.

## Serving data
## Loader strategies

```mermaid
graph LR;
server --> forecast;
forecast --> dataset;
dataset --> index;
dataset --> values;
```
The `loader.type` config key selects how forecast data is held in memory:

Four components are mainly involved in serving data:
| Type | Behaviour |
|---|---|
| `memory` | Downloads the full grid blob at startup into CGo-allocated memory (bypassing GC pressure). Zero I/O at query time. Capped by `max_size_gib`. |
| `blob` | Downloads nothing upfront; issues a byte-range read per query (2 s timeout). |

### server
Use `memory` when latency matters and the dataset fits in RAM; use `blob` otherwise.

Handles incoming grpc requests, including protobuf serialization.
## Native dependencies

### forecast

Determines what is the correct group (and version) to serve data from. Forwards requests to the relevant `dataset` handler.

### dataset

Each object of type `dataset.Dataset` serves data for a single area/version. They maintain a list of grids for its area. A grid is a unique collection of latitude/longitude pairs within a single area. They exist because different parameters may have different grid resolutions.

Handles requests for a given latitute/longitude pair. For each grid, lookup the correct index from `index`, and find relevant data from `values`.

### index

Handles lookup from latitude/longitude to a grid index.

### values

A collection of all data having the same area and grid id.

Provides a `Reader` interface, for looking up data with a given index. The index is provided by the `geo` component. There are several implementations of this interface.

## Loading data

`forecast` component contains a function, `Forecast.update`, that is called periodically in a goroutine. It checks a blob store for updates, and loads data if needed, by calling `dataset.Download`.

## Other modules

### internal.health

Provides grpc healthcheck, meant for kubernetes readiness probe.

### pointdata

Defines internal data format. Its placement reflects that several modules in various places in the hierarchy needs access to this.
The geographic index uses [s2geometry](https://github.com/google/s2geometry) and [PROJ](https://proj.org/) via CGo. These are pre-installed in the devcontainer; building outside it requires both libraries.
11 changes: 11 additions & 0 deletions rawdataforecaster/cmd/lookup/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,9 @@ import (
func main() {
latitude := flag.Float64("lat", 59, "latitude to query")
longitude := flag.Float64("lon", 11, "longitude to query")
altitude := flag.Float64("altitude", -1, "altitude to query; set to 0 for sea level, default -1 means not set")
address := flag.String("address", "localhost:5052", "Server to connect to")
parameter := flag.String("parameter", "", "Only show data for the given parameter")
flag.Parse()

// Set up a connection to the server.
Expand All @@ -33,6 +35,12 @@ func main() {
Latitude: float32(*latitude),
Longitude: float32(*longitude),
}
if *altitude != -1 {
request.Altitude = &internalprotocol.Altitude{
Override: true,
Value: float32(*altitude),
}
}

forecast, err := c.GetForecast(ctx, &request)
if err != nil {
Expand All @@ -46,6 +54,9 @@ func main() {
fmt.Printf("updated at: %v\n", forecast.ForecastMeta.UpdatedAt.AsTime())
fmt.Printf("next update: %v\n", forecast.ForecastMeta.NextUpdate.AsTime())
for _, meta := range forecast.ParameterMeta {
if *parameter != "" && meta.Parameter != *parameter {
continue
}
fmt.Printf("%s (%s):\n", meta.Parameter, meta.Units)
for i, t := range meta.Times {
value := forecast.Data[i+int(meta.SliceFrom)]
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,10 @@ type indexID struct {
GridID string
}

// idCache maps a (area, version, gridID) triple to the MD5 checksum of its lat/lon data.
// checkSumCache maps that checksum to a shared GeoMap (the s2geometry spatial index) with
// a reference count. Together they allow datasets with identical grid geometry to reuse
// the same GeoMap across version updates, avoiding an expensive rebuild.
var idCache map[indexID]string
var checkSumCache map[string]*cachedMaps
var cacheMutex sync.Mutex
Expand Down
Loading