[CONFIG-326][stage][prod] Save download metrics and have reflector emitting it (#121)

* pool vectors udpport from download script

* address feedbacks

* address feedback

* address feedback

* remove netcat timeout

* unblock

* see what's blocking

* fix

* fix

* fix

* fix

* TIL the condition on the while loop is only tested at the beginning of each time through that loop

* fix

* check if there is gap

* don't emit if already present

* experiment

* write downloading time into a file

* fix

* address feedback

* address feedbacks

* emit extra metric after reflector finished downloading the snapshot via s3 client

* surface log

* fix test

* fix

* address feedbacks

* address feedbacks

* explicit flushing stats

* fix

* address feedbacks

* add test for emitMetricFromFile

* fix panic

* fix panic

* fix

* refactor

* run in parallel

---------

Co-authored-by: Hongyu Zhou <[email protected]>

Author: zhou-hongyu

.gitignore

@@ -4,3 +4,4 @@
 /bin
 .coverprofile
 .vscode
+.idea

pkg/reflector/download.go

@@ -71,6 +71,7 @@ func (d *S3Downloader) DownloadTo(w io.Writer) (n int64, err error) {
 	if compressedSize != nil {
 		events.Log("LDB inflated %d -> %d bytes", *compressedSize, n)
 	}
+
 	return
 }
 

pkg/reflector/reflector.go

@@ -1,9 +1,11 @@
 package reflector
 
 import (
+	"bytes"
 	"context"
 	"database/sql"
 	"encoding/base64"
+	"encoding/json"
 	"fmt"
 	"io"
 	"net/url"
@@ -25,6 +27,7 @@ import (
 	"github.com/segmentio/errors-go"
 	"github.com/segmentio/events/v2"
 	_ "github.com/segmentio/events/v2/log" // lets events actually log
+
 	"github.com/segmentio/stats/v4"
 )
 
@@ -73,6 +76,12 @@ type ReflectorConfig struct {
 	BusyTimeoutMS     int                      // optional
 }
 
+type DownloadMetric struct {
+	StartTime  int    `json:"startTime,omitempty"`
+	Downloaded string `json:"downloaded"`
+	Compressed string `json:"compressed"`
+}
+
 type starter interface {
 	Start(ctx context.Context)
 }
@@ -175,6 +184,13 @@ func ReflectorFromConfig(config ReflectorConfig) (*Reflector, error) {
 
 	events.Log("Max known ledger sequence: %{seq}d", maxKnownSeq)
 
+	path := "/var/spool/ctlstore/metrics.json"
+	err = emitMetricFromFile(path)
+	if err != nil {
+		events.Log("Failed to emit metric from file", err)
+	}
+	events.Log("Successfully emitted metric from file")
+
 	// TODO: check Upstream fields
 
 	stop := make(chan struct{})
@@ -280,6 +296,45 @@ func ReflectorFromConfig(config ReflectorConfig) (*Reflector, error) {
 	}, nil
 }
 
+func emitMetricFromFile(path string) error {
+	if _, err := os.Stat(path); err != nil {
+		switch {
+		case os.IsNotExist(err):
+			return nil
+		default:
+			return err
+		}
+	}
+
+	metricsFile, err := os.Open(path)
+	if err != nil {
+		return err
+	}
+
+	b, err := io.ReadAll(metricsFile)
+	if err != nil {
+		return err
+	}
+
+	var dm DownloadMetric
+
+	d := json.NewDecoder(bytes.NewReader(b))
+	d.DisallowUnknownFields()
+	err = d.Decode(&dm)
+	if err != nil {
+		return err
+	}
+
+	stats.Observe("init_snapshot_download_time", dm.StartTime, stats.T("downloaded", dm.Downloaded), stats.T("compressed", dm.Compressed))
+
+	defer func() {
+		metricsFile.Close()
+		os.Remove(path)
+	}()
+
+	return nil
+}
+
 func (r *Reflector) Start(ctx context.Context) error {
 	events.Log("Starting Reflector.")
 	go r.ledgerMonitor.Start(ctx)

pkg/reflector/reflector_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql"
 	"encoding/base64"
+	"github.com/stretchr/testify/assert"
 	"io/ioutil"
 	"os"
 	"path/filepath"
@@ -16,6 +17,7 @@ import (
 	"github.com/segmentio/ctlstore/pkg/ldb"
 	"github.com/segmentio/ctlstore/pkg/ldbwriter"
 	"github.com/segmentio/ctlstore/pkg/ledger"
+	"github.com/segmentio/errors-go"
 	"github.com/segmentio/events/v2"
 	"github.com/stretchr/testify/require"
 )
@@ -220,3 +222,76 @@ func TestReflector(t *testing.T) {
 	err = reflector.Close()
 	require.NoError(t, err)
 }
+
+func TestEmitMetricFromFile(t *testing.T) {
+	for _, test := range []struct {
+		name     string
+		fileName string
+		extra    string
+		content  string
+		perm     int
+		err      error
+	}{
+		{
+			"file does not exist doesn't return error",
+			"1.jso",
+			"n",
+			"{\"startTime\": \"6\", \"downloaded\": \"true\", \"compressed\": \"false\"}",
+			0664,
+			nil,
+		},
+		{
+			"file exist but unable to open",
+			"2.json",
+			"",
+			"{\"startTime\": 6, \"downloaded\": \"true\", \"compressed\": \"false\"}",
+			064,
+			errors.New("permission denied"),
+		},
+		{
+			"invalid character",
+			"3.json",
+			"",
+			"{\"startTime\": \"6, \"downloaded\": \"true\", \"compressed\": \"false\"}",
+			0664,
+			errors.New("invalid character"),
+		},
+		{
+			"invalid key",
+			"4.json",
+			"",
+			"{\"start\": \"6\", \"downloaded\": \"true\", \"compressed\": \"false\"}",
+			0664,
+			errors.New("unknown field"),
+		},
+		{
+			"valid content",
+			"5.json",
+			"",
+			"{\"startTime\": 6, \"downloaded\": \"true\", \"compressed\": \"false\"}",
+			0664,
+			nil,
+		},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			t.Parallel()
+			tempdir := t.TempDir()
+			f, err := os.CreateTemp(tempdir, test.fileName)
+			assert.NoError(t, err)
+
+			_, err = f.Write([]byte(test.content))
+			assert.NoError(t, err)
+
+			err = os.Chmod(f.Name(), os.FileMode(test.perm))
+			assert.NoError(t, err)
+
+			err = emitMetricFromFile(f.Name() + test.extra)
+
+			if test.err == nil {
+				require.NoError(t, err)
+			} else {
+				require.Contains(t, err.Error(), test.err.Error())
+			}
+		})
+	}
+}

scripts/download.sh

@@ -4,10 +4,10 @@ set -eo pipefail
 
 CTLSTORE_BOOTSTRAP_URL=$1
 CONCURRENCY=${2:-20}
-STATS_IP=$3
-STATS_PORT=${4:-8125}
+DOWNLOADED="false"
+COMPRESSED="false"
+METRICS="/var/spool/ctlstore/metrics.json"
 
-TAGS="downloaded:false"
 START=$(date +%s)
 END=$(date +%s)
 if [ ! -f /var/spool/ctlstore/ldb.db ]; then
@@ -18,22 +18,19 @@ if [ ! -f /var/spool/ctlstore/ldb.db ]; then
   cd /var/spool/ctlstore
   s5cmd -r 0 --log debug cp --concurrency $CONCURRENCY $CTLSTORE_BOOTSTRAP_URL .
 
-  TAGS="downloaded:true"
+  DOWNLOADED="true"
   if [[ ${CTLSTORE_BOOTSTRAP_URL: -2} == gz ]]; then
     echo "Decompressing"
     pigz -d snapshot.db.gz
-    TAGS="$TAGS,compressed:true"
+    COMPRESSED="true"
   fi
 
-  TAGS="$TAGS,concurrency:$CONCURRENCY"
-
   mv snapshot.db ldb.db
   END=$(date +%s)
   echo "ldb.db ready in $(($END - $START)) seconds"
 else
   echo "Snapshot already present"
 fi
 
-if [ ! -z "$STATS_IP" ]; then
-  echo -n "ctlstore.reflector.init_snapshot_download_time:$(($END - $START))|h|#$TAGS" | nc -u -w1 $NODE_IP $STATS_PORT
-fi
+echo "{\"startTime\": $(($END - $START)), \"downloaded\": \"$DOWNLOADED\", \"compressed\": \"$COMPRESSED\"}" > $METRICS
+cat $METRICS