Saving work

cmd/accumulated/cmd_migrate.go

@@ -226,12 +226,11 @@ func migrateCfg(cfg *run.Config, cvc *run.CoreValidatorConfiguration, dir string
 		return fmt.Errorf("migration of non-default [api] values not yet supported")
 	}
 
-	if old.Accumulate.Logging.EnableLoki ||
-		old.Accumulate.Logging.LokiUsername != "" ||
-		old.Accumulate.Logging.LokiPassword != "" ||
-		old.Accumulate.Logging.LokiUrl != "" {
+	if old.Accumulate.Logging.EnableLoki &&
+		(old.Accumulate.Logging.LokiUsername != "" ||
+			old.Accumulate.Logging.LokiPassword != "" ||
+			old.Accumulate.Logging.LokiUrl != "") {
 		cfg.Logging.Loki = &run.LokiLogging{
-			Enable:   old.Accumulate.Logging.EnableLoki,
 			Url:      old.Accumulate.Logging.LokiUrl,
 			Username: old.Accumulate.Logging.LokiUsername,
 			Password: old.Accumulate.Logging.LokiPassword,

cmd/accumulated/run/config.yml

@@ -63,8 +63,6 @@ LoggingRule:
 LokiLogging:
   non-binary: true
   fields:
-    - name: Enable
-      type: bool
     - name: Url
       type: string
     - name: Username
@@ -84,6 +82,26 @@ Instrumentation:
     - name: PprofListen
       type: p2p.Multiaddr
       marshal-as: union
+    - name: Push
+      type: PushMetrics
+      marshal-as: reference
+      pointer: true
+
+PushMetrics:
+  non-binary: true
+  fields:
+    - name: Url
+      type: string
+    - name: Username
+      type: string
+    - name: Password
+      type: string
+    - name: WalDirectory
+      type: string
+    - name: ScrapeInterval
+      type: duration
+    - name: RemoteFlushDeadline
+      type: duration
 
 Monitor:
   non-binary: true

cmd/accumulated/run/instrumentation.go

@@ -32,14 +32,16 @@ func (i *Instrumentation) start(inst *Instance) error {
 		return errors.UnknownError.WithFormat("listen: %w", err)
 	}
 
-	if i.Monitoring == nil {
-		i.Monitoring = new(Monitor)
-	}
 	err = i.Monitoring.start(inst)
 	if err != nil {
 		return errors.UnknownError.WithFormat("monitoring: %w", err)
 	}
 
+	err = i.Push.start(inst)
+	if err != nil {
+		return errors.UnknownError.WithFormat("push: %w", err)
+	}
+
 	return nil
 }
 
@@ -93,6 +95,10 @@ func (i *Instrumentation) startPprof(inst *Instance) error {
 }
 
 func (m *Monitor) start(inst *Instance) error {
+	if m == nil {
+		m = new(Monitor)
+	}
+
 	setDefaultPtr(&m.ProfileMemory, false)           // Enabled      = false
 	setDefaultPtr(&m.MemoryPollingRate, time.Minute) // Polling rate = every minute
 	setDefaultPtr(&m.AllocRateTrigger, 50<<20)       // Trigger rate = 50 MiB/s

cmd/accumulated/run/logging.go

@@ -44,7 +44,7 @@ func (l *Logging) newHandler(inst *Instance, out io.Writer) (slog.Handler, error
 		return nil, errors.BadRequest.WithFormat("log format %q is not supported", l.Format)
 	}
 
-	if l.Loki != nil && l.Loki.Enable {
+	if l.Loki != nil || l.Loki.Url == "" {
 		out2, err := l.Loki.start(inst)
 		if err != nil {
 			return nil, errors.UnknownError.WithFormat("Loki: %w", err)

cmd/accumulated/run/push_metrics.go

@@ -0,0 +1,72 @@
+// Copyright 2024 The Accumulate Authors
+//
+// Use of this source code is governed by an MIT-style
+// license that can be found in the LICENSE file or at
+// https://opensource.org/licenses/MIT.
+
+package run
+
+import (
+	"os"
+	"path/filepath"
+	"sync"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/prometheus/scrape"
+	"github.com/prometheus/prometheus/storage/remote"
+	"gitlab.com/accumulatenetwork/accumulate/pkg/errors"
+	"golang.org/x/exp/slog"
+)
+
+func (p *PushMetrics) start(inst *Instance) error {
+	if p == nil || p.Url == "" {
+		return nil
+	}
+
+	setDefaultVal(&p.WalDirectory, filepath.Join(os.TempDir(), "prom-push-wal"))
+	setDefaultVal(&p.ScrapeInterval, time.Minute)
+	setDefaultVal(&p.RemoteFlushDeadline, time.Minute)
+
+	// NOTE: Local querying (read-write storage instead of write-only storage)
+	// would require a local TSDB
+
+	logger := log4prom{*inst.logger.With("module", "metrics")}
+	scrapeMgr := new(deferred[*scrape.Manager])
+	storage := remote.NewWriteStorage(logger, prometheus.DefaultRegisterer, p.WalDirectory, p.RemoteFlushDeadline, scrapeMgr)
+
+	_ = storage
+	return nil
+}
+
+type log4prom struct {
+	slog slog.Logger
+}
+
+func (l log4prom) Log(keyvals ...interface{}) error {
+	l.slog.Info("", keyvals...)
+	return nil
+}
+
+type deferred[T any] struct {
+	mu    sync.RWMutex
+	val   T
+	ready bool
+}
+
+func (d *deferred[T]) Set(v T) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	d.val, d.ready = v, true
+}
+
+func (d *deferred[T]) Get() (T, error) {
+	d.mu.RLock()
+	defer d.mu.RUnlock()
+	if !d.ready {
+		var z T
+		return z, errors.NotReady
+	}
+	return d.val, nil
+
+}