a minor change

Signed-off-by: Genady Gurevich <genadyg@il.ibm.com>

Author: gengur

common/utils/monitoring/monitor.go

@@ -0,0 +1,102 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package monitoring
+
+import (
+	"context"
+	"fmt"
+	"sync/atomic"
+	"time"
+
+	"github.com/hyperledger/fabric-x-orderer/common/types"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+type MonitoringMetric struct {
+	Interval time.Duration
+	Value    *uint64
+	Labels   []string
+}
+
+func (m *MonitoringMetric) monitorBuilder(counter prometheus.Counter) func(ctx context.Context) {
+	return func(ctx context.Context) {
+		ticker := time.NewTicker(m.Interval)
+		defer ticker.Stop()
+
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				v := atomic.LoadUint64(m.Value)
+				atomic.StoreUint64(m.Value, 0)
+				counter.Add(float64(v))
+			}
+		}
+	}
+}
+
+type Monitor struct {
+	monitor  *Provider
+	logger   types.Logger
+	host     string
+	endpoint Endpoint
+	monitors []func(context.Context)
+	// stop is used to stop the monitoring service
+	stop context.CancelFunc
+}
+
+type MonitorOpts struct {
+	Host string
+	prometheus.CounterOpts
+	Labels  []string
+	Metrics []MonitoringMetric
+}
+
+// NewMonitor creates and returns a new Monitor instance using the provided MonitorOpts and logger.
+// It initializes a metrics provider, sets up a CounterVec with the specified options and labels,
+// and constructs monitor functions for each metric defined in the MonitorOpts.
+// The returned Monitor is configured to track and log metrics as specified.
+//
+// Parameters:
+//   - monitorOpts: Pointer to MonitorOpts containing configuration for metrics, labels, and host.
+//   - logger:      Logger instance for logging within the Monitor.
+//
+// Returns:
+//   - Pointer to the initialized Monitor.
+func NewMonitor(monitorOpts *MonitorOpts, logger types.Logger) *Monitor {
+	provider := NewProvider()
+	counterVec := provider.NewCounterVec(monitorOpts.CounterOpts, monitorOpts.Labels)
+
+	monitors := make([]func(context.Context), len(monitorOpts.Metrics))
+	for i := range monitorOpts.Metrics {
+		monitors[i] = monitorOpts.Metrics[i].monitorBuilder(counterVec.WithLabelValues(monitorOpts.Metrics[i].Labels...))
+	}
+
+	return &Monitor{monitor: provider, host: monitorOpts.Host, logger: logger, monitors: monitors}
+}
+
+func (m *Monitor) Start() {
+	m.endpoint = Endpoint{Host: m.host, Port: 2112}
+	ctx, cancel := context.WithCancel(context.Background())
+	m.stop = cancel
+	serverConfig := ServerConfig{Endpoint: m.endpoint}
+
+	go func() {
+		m.monitor.StartPrometheusServer(ctx, &serverConfig, m.monitors...)
+	}()
+}
+
+func (m *Monitor) Stop() {
+	if m.stop != nil {
+		m.stop()
+	}
+}
+
+func (m *Monitor) Address() string {
+	return fmt.Sprintf("http://%s/metrics", m.endpoint.Address())
+}

common/utils/monitoring/promutil/metric_setter.go

@@ -0,0 +1,58 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package promutil
+
+import (
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// AddToCounter adds a value to a prometheus counter.
+func AddToCounter(c prometheus.Counter, n int) {
+	c.Add(float64(n))
+}
+
+// AddToCounterVec add a value to given labels on a prometheus counter vector.
+func AddToCounterVec(c *prometheus.CounterVec, labels []string, n int) {
+	c.WithLabelValues(labels...).Add(float64(n))
+}
+
+// AddToGauge adds a value to a prometheus gauge.
+func AddToGauge(g prometheus.Gauge, n int) {
+	g.Add(float64(n))
+}
+
+// SubFromGauge subtracts a given value from a prometheus gauge.
+func SubFromGauge(g prometheus.Gauge, n int) {
+	g.Sub(float64(n))
+}
+
+// SetGaugeVec sets a value to given labels on the prometheus gauge vector.
+func SetGaugeVec(c *prometheus.GaugeVec, labels []string, n int) {
+	c.WithLabelValues(labels...).Set(float64(n))
+}
+
+// SetGauge sets a value to a prometheus gauge.
+func SetGauge(queue prometheus.Gauge, n int) {
+	queue.Set(float64(n))
+}
+
+// SetUint64Gauge sets a uint64 value to a prometheus gauge.
+func SetUint64Gauge(queue prometheus.Gauge, n uint64) {
+	queue.Set(float64(n))
+}
+
+// Observe observes a prometheus histogram.
+func Observe(h prometheus.Observer, d time.Duration) {
+	h.Observe(d.Seconds())
+}
+
+// ObserveSize observes a prometheus histogram size.
+func ObserveSize(h prometheus.Observer, size int) {
+	h.Observe(float64(size))
+}

common/utils/monitoring/provider.go

@@ -0,0 +1,160 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package monitoring
+
+import (
+	"context"
+	"net/http"
+	"net/url"
+	"time"
+
+	"github.com/hyperledger/fabric-lib-go/common/flogging"
+	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"golang.org/x/sync/errgroup"
+)
+
+const (
+	scheme         = "http://"
+	metricsSubPath = "/metrics"
+)
+
+// Provider is a prometheus metrics provider.
+type Provider struct {
+	registry *prometheus.Registry
+	url      string
+}
+
+var logger = flogging.MustGetLogger("monitoring")
+
+// NewProvider creates a new prometheus metrics provider.
+func NewProvider() *Provider {
+	return &Provider{registry: prometheus.NewRegistry()}
+}
+
+// StartPrometheusServer starts a prometheus server.
+// It also starts the given monitoring methods. Their context will cancel once the server is cancelled.
+// This method returns once the server is shutdown and all monitoring methods returns.
+func (p *Provider) StartPrometheusServer(
+	ctx context.Context, serverConfig *ServerConfig, monitor ...func(context.Context),
+) error {
+	logger.Debugf("Creating prometheus server")
+	mux := http.NewServeMux()
+	mux.Handle(
+		metricsSubPath,
+		promhttp.HandlerFor(
+			p.Registry(),
+			promhttp.HandlerOpts{
+				Registry: p.Registry(),
+			},
+		),
+	)
+	server := &http.Server{
+		ReadTimeout: 30 * time.Second,
+		Handler:     mux,
+	}
+
+	l, err := serverConfig.Listener()
+	if err != nil {
+		return errors.Wrap(err, "failed to start prometheus server")
+	}
+	defer l.Close()
+
+	p.url, err = MakeMetricsURL(l.Addr().String())
+	if err != nil {
+		return errors.Wrap(err, "failed formatting URL")
+	}
+
+	g, gCtx := errgroup.WithContext(ctx)
+	g.Go(func() error {
+		logger.Infof("Prometheus serving on URL: %s", p.url)
+		defer logger.Info("Prometheus stopped serving")
+		return server.Serve(l)
+	})
+
+	// The following ensures the method does not return before all monitor methods return.
+	for _, m := range monitor {
+		g.Go(func() error {
+			m(gCtx)
+			return nil
+		})
+	}
+
+	// The following ensures the method does not return before the close procedure is complete.
+	stopAfter := context.AfterFunc(ctx, func() {
+		g.Go(func() error {
+			if errClose := server.Close(); err != nil {
+				return errors.Wrap(errClose, "failed to close prometheus server")
+			}
+			return nil
+		})
+	})
+	defer stopAfter()
+
+	if err = g.Wait(); !errors.Is(err, http.ErrServerClosed) {
+		return errors.Wrap(err, "prometheus server stopped with an error")
+	}
+	return nil
+}
+
+// URL returns the prometheus server URL.
+func (p *Provider) URL() string {
+	return p.url
+}
+
+// NewCounter creates a new prometheus counter.
+func (p *Provider) NewCounter(opts prometheus.CounterOpts) prometheus.Counter {
+	c := prometheus.NewCounter(opts)
+	p.registry.MustRegister(c)
+	return c
+}
+
+// NewCounterVec creates a new prometheus counter vector.
+func (p *Provider) NewCounterVec(opts prometheus.CounterOpts, labels []string) *prometheus.CounterVec {
+	cv := prometheus.NewCounterVec(opts, labels)
+	p.registry.MustRegister(cv)
+	return cv
+}
+
+// NewGauge creates a new prometheus gauge.
+func (p *Provider) NewGauge(opts prometheus.GaugeOpts) prometheus.Gauge {
+	g := prometheus.NewGauge(opts)
+	p.registry.MustRegister(g)
+	return g
+}
+
+// NewGaugeVec creates a new prometheus gauge vector.
+func (p *Provider) NewGaugeVec(opts prometheus.GaugeOpts, labels []string) *prometheus.GaugeVec {
+	gv := prometheus.NewGaugeVec(opts, labels)
+	p.registry.MustRegister(gv)
+	return gv
+}
+
+// NewHistogram creates a new prometheus histogram.
+func (p *Provider) NewHistogram(opts prometheus.HistogramOpts) prometheus.Histogram {
+	h := prometheus.NewHistogram(opts)
+	p.registry.MustRegister(h)
+	return h
+}
+
+// NewHistogramVec creates a new prometheus histogram vector.
+func (p *Provider) NewHistogramVec(opts prometheus.HistogramOpts, labels []string) *prometheus.HistogramVec {
+	hv := prometheus.NewHistogramVec(opts, labels)
+	p.registry.MustRegister(hv)
+	return hv
+}
+
+// Registry returns the prometheus registry.
+func (p *Provider) Registry() *prometheus.Registry {
+	return p.registry
+}
+
+// MakeMetricsURL construct the Prometheus metrics URL.
+func MakeMetricsURL(address string) (string, error) {
+	return url.JoinPath(scheme, address, metricsSubPath)
+}

common/utils/monitoring/server_util.go

@@ -0,0 +1,73 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package monitoring
+
+import (
+	"net"
+	"os/exec"
+	"strconv"
+	"strings"
+
+	"github.com/pkg/errors"
+)
+
+const grpcProtocol = "tcp"
+
+type Endpoint struct {
+	Host string
+	Port int
+}
+
+// Address returns a string representation of the endpoint's address.
+func (e *Endpoint) Address() string {
+	return net.JoinHostPort(e.Host, strconv.Itoa(e.Port))
+}
+
+type ServerConfig struct {
+	Endpoint             Endpoint
+	preAllocatedListener net.Listener
+}
+
+// Listener instantiate a [net.Listener] and updates the config port with the effective port.
+func (c *ServerConfig) Listener() (net.Listener, error) {
+	if c.preAllocatedListener != nil {
+		return c.preAllocatedListener, nil
+	}
+	listener, err := net.Listen(grpcProtocol, c.Endpoint.Address())
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to listen")
+	}
+
+	addr := listener.Addr()
+	tcpAddress, ok := addr.(*net.TCPAddr)
+	if !ok {
+		return nil, errors.New(strings.Join([]string{"failed to cast to TCP address", listener.Close().Error()}, "; "))
+	}
+	c.Endpoint.Port = tcpAddress.Port
+
+	logger.Infof("Listening on: %s://%s", grpcProtocol, c.Endpoint.Address())
+	return listener, nil
+}
+
+// PreAllocateListener is used to allocate a port and bind to ahead of the server initialization.
+// It stores the listener object internally to be reused on subsequent calls to Listener().
+func (c *ServerConfig) PreAllocateListener() (net.Listener, error) {
+	listener, err := c.Listener()
+	if err != nil {
+		return nil, err
+	}
+	c.preAllocatedListener = listener
+	return listener, nil
+}
+
+func FQDN() (string, error) {
+	out, err := exec.Command("hostname", "--fqdn").Output()
+	if err != nil {
+		return "", err
+	}
+	return string(out), nil
+}