Flush strategy

.dockerignore

@@ -1,6 +1,7 @@
 deploy/
 build/
 helm/
+toolbox/
 .git/
 .github/
 .vscode/

cmd/maps-tile-uploader/defaultConfig.yaml

@@ -3,6 +3,11 @@ minZoom: 0
 planetMaxZoom: 9
 maxTileDownloadRetries: 100
 maxTileDownloadRetryDelaySec: 10
+flushStrategy:
+  # maxFlushInterval: 1h
+  flushOnDetailedColumnFinished: false
+  flushOnColumnFinished: false
+  flushOnZLayerFinished: true
 detailedRegions:
   - name: europe-luxembourg
     geoBBox:

cmd/maps-tile-uploader/flush_strategy.go

@@ -0,0 +1,92 @@
+package main
+
+import (
+	"context"
+	"log/slog"
+	"time"
+
+	"github.com/cinode/go/pkg/cinodefs"
+)
+
+type FlushStrategy interface {
+	FlushOpportunity(ctx context.Context) error
+	ColumnFinished(ctx context.Context, isDetailedRegion bool) error
+	ZLayerFinished(ctx context.Context) error
+	ProcessFinished(ctx context.Context) error
+}
+
+func NewFlushStrategy(
+	fs cinodefs.FS,
+	cfg FlushStrategyConfig,
+	timeSource func() time.Time,
+	log *slog.Logger,
+) FlushStrategy {
+	return &flushStrategy{
+		fs:         fs,
+		cfg:        cfg,
+		timeSource: timeSource,
+		log:        log,
+	}
+}
+
+type FlushStrategyConfig struct {
+	MaxFlushInterval              *time.Duration `yaml:"maxFlushInterval,omitempty"`
+	FlushOnDetailedColumnFinished bool           `yaml:"flushOnDetailedColumnFinished"`
+	FlushOnColumnFinished         bool           `yaml:"flushOnColumnFinished"`
+	FlushOnZLayerFinished         bool           `yaml:"flushOnZLayerFinished"`
+}
+
+type flushStrategy struct {
+	fs            cinodefs.FS
+	lastFlushTime time.Time
+	cfg           FlushStrategyConfig
+	timeSource    func() time.Time
+	log           *slog.Logger
+}
+
+func (f *flushStrategy) flush(ctx context.Context, reason string) error {
+	f.log.InfoContext(ctx, "Flushing filesystem", "reason", reason)
+	if err := f.fs.Flush(ctx); err != nil {
+		return err
+	}
+
+	f.lastFlushTime = f.timeSource()
+
+	return nil
+}
+
+func (f *flushStrategy) FlushOpportunity(ctx context.Context) error {
+	if f.cfg.MaxFlushInterval == nil {
+		return nil
+	}
+
+	if f.timeSource().Sub(f.lastFlushTime) < *f.cfg.MaxFlushInterval {
+		return nil
+	}
+
+	return f.flush(ctx, "maxFlushInterval reached")
+}
+
+func (f *flushStrategy) ColumnFinished(ctx context.Context, isDetailedRegion bool) error {
+	if isDetailedRegion && f.cfg.FlushOnDetailedColumnFinished {
+		return f.flush(ctx, "detailed column finished")
+	}
+
+	if f.cfg.FlushOnColumnFinished {
+		return f.flush(ctx, "column finished")
+	}
+
+	return nil
+}
+
+func (f *flushStrategy) ZLayerFinished(ctx context.Context) error {
+	if f.cfg.FlushOnZLayerFinished {
+		return f.flush(ctx, "zoom layer finished")
+	}
+
+	return nil
+}
+
+func (f *flushStrategy) ProcessFinished(ctx context.Context) error {
+	return f.flush(ctx, "process finished")
+}

cmd/maps-tile-uploader/main.go

@@ -98,15 +98,19 @@ func main() {
 		fmt.Printf("  WriterInfo: %s\n", golang.Must(fs.RootWriterInfo(ctx)))
 	}
 
+	log := slog.Default()
+
 	gen := tilesGenerator{
-		cfg: cfg,
-		fs:  fs,
-		log: slog.Default(),
+		cfg:   cfg,
+		fs:    fs,
+		log:   log,
+		flush: NewFlushStrategy(fs, cfg.FlushStrategy, time.Now, log),
 	}
 
 	err = gen.Process(ctx)
 	if err != nil {
-		log.Fatal(err)
+		log.ErrorContext(ctx, "Failed to process tiles", "err", err)
+		os.Exit(1)
 	}
 }
 
@@ -123,12 +127,14 @@ type Config struct {
 	MaxTileDownloadRetries       int                    `yaml:"maxTileDownloadRetries"`
 	MaxTileDownloadRetryDelaySec int                    `yaml:"maxTileDownloadRetryDelaySec"`
 	DetailedRegions              []DetailedRegionConfig `yaml:"detailedRegions"`
+	FlushStrategy                FlushStrategyConfig    `yaml:"flushStrategy"`
 }
 
 type tilesGenerator struct {
-	cfg Config
-	fs  cinodefs.FS
-	log *slog.Logger
+	cfg   Config
+	fs    cinodefs.FS
+	log   *slog.Logger
+	flush FlushStrategy
 }
 
 func (t *tilesGenerator) fetchTile(
@@ -162,6 +168,10 @@ func (t *tilesGenerator) fetchTile(
 	for retry := 0; ctx.Err() == nil; retry++ {
 		log := log.With("url", url, "retry", retry)
 
+		if err := t.flush.FlushOpportunity(ctx); err != nil {
+			return fmt.Errorf("failed to flush the filesystem: %w", err)
+		}
+
 		log.InfoContext(ctx, "Fetching tile started")
 
 		resp, err := http.Get(url)
@@ -211,6 +221,10 @@ func (t *tilesGenerator) fetchTile(
 
 		log.InfoContext(ctx, "Tile uploaded to cinode", "bn", ep.BlobName().String())
 
+		if err := t.flush.FlushOpportunity(ctx); err != nil {
+			return fmt.Errorf("failed to flush the filesystem: %w", err)
+		}
+
 		return nil
 	}
 
@@ -250,6 +264,11 @@ func (t *tilesGenerator) genXLayer(
 			return err
 		}
 	}
+
+	if err := t.flush.ColumnFinished(ctx, true); err != nil {
+		return fmt.Errorf("failed to flush the filesystem: %w", err)
+	}
+
 	return nil
 }
 
@@ -288,13 +307,12 @@ func (t *tilesGenerator) genZLayer(
 		if err != nil {
 			return err
 		}
+	}
 
-		// For region-based z layer, flush once every column for better persistency and faster results
-		err = t.fs.Flush(ctx)
-		if err != nil {
-			return fmt.Errorf("failed to flush the filesystem: %w", err)
-		}
+	if err := t.flush.ZLayerFinished(ctx); err != nil {
+		return fmt.Errorf("failed to flush the filesystem: %w", err)
 	}
+
 	return nil
 }
 
@@ -310,6 +328,7 @@ func (t *tilesGenerator) genZLayerNoConstraints(
 			return err
 		}
 	}
+
 	return nil
 }
 
@@ -329,6 +348,11 @@ func (t *tilesGenerator) genXLayerNoConstraints(
 			return err
 		}
 	}
+
+	if err := t.flush.ColumnFinished(ctx, false); err != nil {
+		return fmt.Errorf("failed to flush the filesystem: %w", err)
+	}
+
 	return nil
 }
 
@@ -344,13 +368,11 @@ func (t *tilesGenerator) Process(ctx context.Context) error {
 		if err != nil {
 			return err
 		}
-		err = t.fs.Flush(ctx)
-		if err != nil {
-			return fmt.Errorf("failed to flush the filesystem: %w", err)
-		}
+	}
+
+	if err := t.flush.ProcessFinished(ctx); err != nil {
+		return fmt.Errorf("failed to flush the filesystem: %w", err)
 	}
 
 	return nil
 }
-
-// https://github.com/openstreetmap/mod_tile/pull/263#issuecomment-1006034286

helm/osm-machinery/templates/cinode-upload-config-configmap.yaml

@@ -9,6 +9,11 @@ data:
     planetMaxZoom: {{ $.Values.cinodeUpload.planetMaxZoom | int }}
     maxTileDownloadRetries: {{ $.Values.cinodeUpload.maxTileDownloadRetries | int }}
     maxTileDownloadRetryDelaySec: {{ $.Values.cinodeUpload.maxTileDownloadRetryDelaySec | int }}
+    flushStrategy:
+      maxFlushInterval: 1h
+      flushOnDetailedColumnFinished: true
+      flushOnColumnFinished: true
+      flushOnZLayerFinished: true
     detailedRegions:
     {{- range $.Values.regions }}
     {{-   if .enabled }}

toolbox/minikube-test.sh

@@ -0,0 +1,48 @@
+#!/bin/bash
+
+set -euo pipefail
+
+cd "$(dirname "$0")/.."
+
+if ! minikube status | grep -q "Running"; then
+    echo "Minikube is not running"
+    exit 1
+fi
+
+eval "$(minikube docker-env)"
+
+if git diff --quiet; then
+    TAGNAME="devel-$(git rev-parse --short HEAD)"
+else
+    TAGNAME="devel-$(git rev-parse --short HEAD)-dirty-$(date +%s)"
+fi
+
+for image in \
+    maps-tile-uploader \
+    maps-tile-server \
+; do
+    docker build -t cinode/$image:$TAGNAME -f build/docker/Dockerfile.${image} .
+done
+
+VALUES_CONTENT="---
+cinodeUpload:
+    image:
+        tag: $TAGNAME
+        registry: docker.io
+        repository: cinode/maps-tile-uploader
+        pullPolicy: Never
+tileServer:
+    image:
+        tag: $TAGNAME
+        registry: docker.io
+        repository: cinode/maps-tile-server
+        pullPolicy: Never
+"
+
+helm \
+    upgrade --install \
+    cinode-maps-tile-server \
+    ./helm/osm-machinery \
+    --kube-context minikube \
+    --values ./helm/osm-machinery/values.yaml \
+    --values <( echo "$VALUES_CONTENT" )