tstor/daemons/torrent/metrics.go

package torrent

import (
	"context"
	"encoding/base64"

	"github.com/anacrolix/dht/v2"
	"github.com/anacrolix/torrent"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/metric"
)

func registerTorrentMetrics(client *torrent.Client) error {
	meterTotalPeers, _ := meter.Int64ObservableGauge("torrent.peers.total")
	meterActivePeers, _ := meter.Int64ObservableGauge("torrent.peers.active")
	meterSeeders, _ := meter.Int64ObservableGauge("torrent.seeders")
	meterDownloaded, _ := meter.Int64ObservableGauge("torrent.downloaded", metric.WithUnit("By"))
	meterIO, _ := meter.Int64ObservableGauge("torrent.io", metric.WithUnit("By"))
	meterLoaded, _ := meter.Int64ObservableGauge("torrent.loaded")

	_, err := meter.RegisterCallback(func(ctx context.Context, o metric.Observer) error {
		o.ObserveInt64(meterLoaded, int64(len(client.Torrents())))

		for _, v := range client.Torrents() {
			as := attribute.NewSet(
				attribute.String("infohash", v.InfoHash().HexString()),
				attribute.String("name", v.Name()),
				attribute.Int64("size", v.Length()),
			)
			stats := v.Stats()
			o.ObserveInt64(meterTotalPeers, int64(stats.TotalPeers), metric.WithAttributeSet(as))
			o.ObserveInt64(meterActivePeers, int64(stats.ActivePeers), metric.WithAttributeSet(as))
			o.ObserveInt64(meterSeeders, int64(stats.ConnectedSeeders), metric.WithAttributeSet(as))
			o.ObserveInt64(meterIO, stats.BytesRead.Int64(), metric.WithAttributeSet(as), metric.WithAttributes(attribute.String("direction", "download")))
			o.ObserveInt64(meterIO, stats.BytesWritten.Int64(), metric.WithAttributeSet(as), metric.WithAttributes(attribute.String("direction", "upload")))
			o.ObserveInt64(meterDownloaded, v.BytesCompleted(), metric.WithAttributeSet(as))
		}

		return nil
	}, meterTotalPeers, meterActivePeers, meterSeeders, meterIO, meterDownloaded, meterLoaded)
	if err != nil {
		return err
	}
	return nil
}

func registerDhtMetrics(client *torrent.Client) error {
	meterDhtNodes, _ := meter.Int64ObservableGauge("torrent.dht.nodes")

	_, err := meter.RegisterCallback(func(ctx context.Context, o metric.Observer) error {
		servers := client.DhtServers()
		for _, dhtSrv := range servers {
			stats, ok := dhtSrv.Stats().(dht.ServerStats)
			if !ok {
				continue
			}
			id := dhtSrv.ID()
			as := attribute.NewSet(
				attribute.String("id", base64.StdEncoding.EncodeToString(id[:])),
				attribute.String("address", dhtSrv.Addr().String()),
			)
			o.ObserveInt64(meterDhtNodes, int64(stats.Nodes), metric.WithAttributeSet(as))
		}

		return nil
	}, meterDhtNodes)

	return err
}