Add Bunny CDN Replication Prober (#1)

Application to measure replication delay of edge storage and edge rules of Bunny CDN.

This also includes 2 potentially useful packages:
- Bunny CDN API client
- Generic caching class with single flight updater

.github/workflows/ci.yaml

@@ -4,15 +4,10 @@ jobs:
   build-test:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
-      - uses: actions/cache@v2
+      - uses: actions/checkout@v3
+      - uses: actions/setup-go@v4
         with:
-          path: ~/go/pkg/mod
-          key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
-      - name: Setup Go
-        uses: actions/setup-go@v2
-        with:
-          go-version: '^1.18.0'
+          go-version: '^1.21.0'
       - name: Build
         run: go build ./...
       - name: Test

.gitignore

@@ -1 +1,3 @@
 /rapidsyncer
+/replicationprober
+.envrc

build/Dockerfile → build/rapidsyncer.Dockerfile

@@ -1,4 +1,4 @@
-FROM docker.io/library/golang:1.18 AS build
+FROM docker.io/library/golang:1.21-bullseye AS build
 
 WORKDIR /go/src/app
 

build/replicationprober.Dockerfile

@@ -0,0 +1,16 @@
+FROM docker.io/library/golang:1.21-bullseye AS build
+
+WORKDIR /go/src/app
+
+COPY go.mod ./
+COPY go.sum ./
+RUN go mod download -x
+
+COPY . .
+RUN CGO_ENABLED=0 go build ./cmd/replicationprober
+
+FROM gcr.io/distroless/static-debian11
+WORKDIR /
+USER nonroot:nonroot
+COPY --from=build /go/src/app/replicationprober /
+ENTRYPOINT ["/replicationprober"]

cmd/replicationprober/README.md

@@ -0,0 +1,12 @@
+# Replication Prober
+
+Replication Prober is a tool for testing replication of data in Bunny CDN.
+
+- `RedirectProber`: tests current status of HTTP Redirects for the
+  `{repo}/version.gz` file that are set up using
+  https://github.com/p2004a/bar-repos-bunny-replication-lag-mitigation.
+- `StorageReplicationProber`: writes a special canary file to Bunny Edge Storage
+  and measures after what time the it's available in the configured replication
+  zones.
+
+See `config` struct in `main.go` for the ENV configuration options.

cmd/replicationprober/main.go

@@ -0,0 +1,141 @@
+// SPDX-FileCopyrightText: 2023 Marek Rusinowski
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"net/http"
+	"time"
+
+	"github.com/InfluxCommunity/influxdb3-go/influxdb3"
+	"github.com/caarlos0/env/v9"
+	"github.com/p2004a/spring-rapid-syncer/pkg/bunny"
+	"github.com/p2004a/spring-rapid-syncer/pkg/sfcache"
+)
+
+const UserAgent = "spring-rapid-syncer/prober 1.0"
+
+const versionsGzRepo = "byar"
+const versionsGzFile = "/" + versionsGzRepo + "/versions.gz"
+
+func getExpectedStorageRegions(ctx context.Context, bunnyClient *bunny.Client, storageZone string) ([]string, error) {
+	zone, err := bunnyClient.StorageZoneByName(ctx, storageZone)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get storage zones: %w", err)
+	}
+	regions := zone.ReplicationRegions[:]
+	regions = append(regions, zone.Region)
+	return regions, nil
+}
+
+type config struct {
+	Bunny struct {
+		AccessKey   string
+		StorageZone string
+	} `envPrefix:"BUNNY_"`
+	InfluxDb struct {
+		Url      string
+		Token    string
+		Database string
+	} `envPrefix:"INFLUXDB_"`
+	BaseUrl                        string        `envDefault:"https://repos-cdn.beyondallreason.dev"`
+	MaxRegionDistanceKm            float64       `envDefault:"400"`
+	StorageEdgeMapCacheDuration    time.Duration `envDefault:"24h"`
+	VersionGzCacheDuration         time.Duration `envDefault:"10s"`
+	RefreshReplicationCanaryPeriod time.Duration `envDefault:"5m"`
+	CheckReplicationStatusPeriod   time.Duration `envDefault:"2m"`
+	CheckRedirectStatusPeriod      time.Duration `envDefault:"2m"`
+	EnableStorageReplicationProber bool          `envDefault:"true"`
+	EnableRedirectStatusProber     bool          `envDefault:"true"`
+	Port                           string        `envDefault:"disabled"`
+}
+
+func main() {
+	cfg := config{}
+	if err := env.ParseWithOptions(&cfg, env.Options{
+		RequiredIfNoDef:       true,
+		UseFieldNameByDefault: true,
+	}); err != nil {
+		log.Fatalf("%+v\n", err)
+	}
+
+	bunnyClient := bunny.NewClient(cfg.Bunny.AccessKey)
+
+	ctx := context.Background()
+	bunnyStorageZoneClient, err := bunnyClient.NewStorageZoneClient(ctx, cfg.Bunny.StorageZone)
+	if err != nil {
+		log.Fatalf("Failed to create Bunny storage zone client: %v", err)
+	}
+
+	expectedStorageRegions, err := getExpectedStorageRegions(ctx, bunnyClient, cfg.Bunny.StorageZone)
+	if err != nil {
+		log.Fatalf("Failed to get expected storage regions: %v", err)
+	}
+
+	influxdbClient, err := influxdb3.New(influxdb3.ClientConfig{
+		Host:     cfg.InfluxDb.Url,
+		Token:    cfg.InfluxDb.Token,
+		Database: cfg.InfluxDb.Database,
+	})
+	if err != nil {
+		log.Fatalf("Failed to create InfluxDB client: %v", err)
+	}
+
+	rf := &ReplicatedFetcher{
+		bunny: bunnyClient,
+		storageEdgeMapCache: sfcache.Cache[StorageEdgeMap]{
+			Timeout: cfg.StorageEdgeMapCacheDuration,
+		},
+		maxRegionDistanceKm:    cfg.MaxRegionDistanceKm,
+		expectedStorageRegions: expectedStorageRegions,
+		probeFileUrl:           cfg.BaseUrl + "/empty.txt",
+	}
+	http.HandleFunc("/storageedgemap", rf.HandleStorageEdgeMap)
+
+	srp := &StorageReplicationProber{
+		replicatedFetcher:              rf,
+		bunnyStorageZone:               bunnyStorageZoneClient,
+		influxdbClient:                 influxdbClient,
+		refreshReplicationCanaryPeriod: cfg.RefreshReplicationCanaryPeriod,
+		checkReplicationStatusPeriod:   cfg.CheckRedirectStatusPeriod,
+		canaryFileUrl:                  cfg.BaseUrl + "/replication-canary.txt",
+		canaryFileName:                 "replication-canary.txt",
+	}
+	if cfg.EnableStorageReplicationProber {
+		go srp.startCanaryUpdater(ctx)
+		go srp.startReplicationStatusChecker(ctx)
+	}
+	http.HandleFunc("/replicationstatus", srp.HandleReplicationStatus)
+
+	redirectProber := &RedirectProber{
+		bunny:                   bunnyClient,
+		influxdbClient:          influxdbClient,
+		repo:                    versionsGzRepo,
+		versionGzUrl:            cfg.BaseUrl + versionsGzFile,
+		probePeriod:             cfg.CheckRedirectStatusPeriod,
+		edgeServerRefreshPeriod: time.Hour * 1,
+	}
+	if cfg.EnableRedirectStatusProber {
+		go redirectProber.startProber(ctx)
+	}
+
+	versionGzFetcher := &VersionGzFetcher{
+		replicatedFetcher: rf,
+		versionGzUrl:      cfg.BaseUrl + versionsGzFile,
+		versionsGzCache: sfcache.Cache[[]*ReplicatedFile]{
+			Timeout: cfg.VersionGzCacheDuration,
+		},
+	}
+	http.HandleFunc("/replicationstatus_versiongz", versionGzFetcher.HandleReplicationStatusVersionsGz)
+
+	if cfg.Port == "disabled" {
+		select {}
+	} else {
+		if err := http.ListenAndServe(":"+cfg.Port, nil); err != nil {
+			log.Fatal(err)
+		}
+	}
+}