(manual census reporting) add CLI subcommand and API endpoint for census utilization

Author: sujay-hashicorp

agent/agent.go

@@ -1602,6 +1602,7 @@ func newConsulConfig(runtimeCfg *config.RuntimeConfig, logger hclog.Logger) (*co
 	cfg.Cloud = runtimeCfg.Cloud
 
 	cfg.Reporting.License.Enabled = runtimeCfg.Reporting.License.Enabled
+	cfg.Reporting.SnapshotRetentionTime = runtimeCfg.Reporting.SnapshotRetentionTime
 
 	cfg.ServerRejoinAgeMax = runtimeCfg.ServerRejoinAgeMax
 	cfg.EnableXDSLoadBalancing = runtimeCfg.EnableXDSLoadBalancing

agent/config/config.go

@@ -1000,5 +1000,6 @@ type License struct {
 }
 
 type Reporting struct {
-	License License `mapstructure:"license"`
+	License               License `mapstructure:"license"`
+	SnapshotRetentionTime *string `mapstructure:"snapshot_retention_time" json:"snapshot_retention_time,omitempty"`
 }

agent/config/runtime.go

@@ -1541,7 +1541,8 @@ type LicenseConfig struct {
 }
 
 type ReportingConfig struct {
-	License LicenseConfig
+	License               LicenseConfig
+	SnapshotRetentionTime time.Duration
 }
 
 type AutoConfig struct {

agent/consul/config.go

@@ -17,6 +17,7 @@ import (
 
 	"github.com/hashicorp/consul/agent/checks"
 	consulrate "github.com/hashicorp/consul/agent/consul/rate"
+	"github.com/hashicorp/consul/agent/consul/reporting"
 	hcpconfig "github.com/hashicorp/consul/agent/hcp/config"
 	"github.com/hashicorp/consul/agent/structs"
 	"github.com/hashicorp/consul/internal/gossip/libserf"
@@ -597,6 +598,8 @@ func DefaultConfig() *Config {
 		ServerRejoinAgeMax: 24 * 7 * time.Hour,
 	}
 
+	conf.Reporting.SnapshotRetentionTime = reporting.DefaultSnapshotRetention
+
 	// Increase our reap interval to 3 days instead of 24h.
 	conf.SerfLANConfig.ReconnectTimeout = 3 * 24 * time.Hour
 	conf.SerfWANConfig.ReconnectTimeout = 3 * 24 * time.Hour
@@ -732,5 +735,6 @@ type License struct {
 }
 
 type Reporting struct {
-	License License
+	License               License
+	SnapshotRetentionTime time.Duration
 }

agent/consul/leader.go

@@ -340,9 +340,8 @@ func (s *Server) establishLeadership(ctx context.Context) error {
 		s.startLogVerification(ctx)
 	}
 
-	if s.config.Reporting.License.Enabled && s.reportingManager != nil {
-		s.reportingManager.StartReportingAgent()
-	}
+	// The reporting manager will start the metrics collection but will only report if config.reporting.license.enabled is true
+	s.reportingManager.StartReportingAgent()
 
 	s.logger.Debug("successfully established leadership", "duration", time.Since(start))
 	return nil

agent/consul/reporting/reporting.go

@@ -11,45 +11,75 @@ import (
 	"github.com/hashicorp/consul/agent/structs"
 	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/go-memdb"
+	"github.com/hashicorp/go-retryablehttp"
 )
 
 type ReportingManager struct {
-	logger         hclog.Logger
-	server         ServerDelegate
-	stateProvider  StateDelegate
-	tickerInterval time.Duration
+	logger                 hclog.Logger
+	clusterId              string
+	autoReporting          bool
+	server                 ServerDelegate
+	stateProvider          StateDelegate
+	tickerInterval         time.Duration
+	manualSnapshotInterval time.Duration
+	snapshotRetention      time.Duration
+	customerID             string
 	EntDeps
 	sync.RWMutex
+	manualHTTPClient     *retryablehttp.Client
+	manualServiceAddress string
 }
 
 const (
 	SystemMetadataReportingProcessID = "reporting-process-id"
 	ReportingInterval                = 1 * time.Hour
 )
 
+const (
+	// ManualSnapshotInterval controls how often we persist manual census snapshots.
+	ManualSnapshotInterval = 24 * time.Hour
+	// DefaultSnapshotRetention is the default retention period for manual census snapshots.
+	DefaultSnapshotRetention = 9600 * time.Hour // 400 days
+)
+
 //go:generate mockery --name ServerDelegate --inpackage
 type ServerDelegate interface {
 	GetSystemMetadata(key string) (string, error)
 	SetSystemMetadataKey(key, val string) error
 	IsLeader() bool
+	ApplyCensusRequest(req *structs.CensusRequest) error
 }
 
 type StateDelegate interface {
+	// Metrics methods
 	NodeUsage() (uint64, state.NodeUsage, error)
 	ServiceUsage(ws memdb.WatchSet, tenantUsage bool) (uint64, structs.ServiceUsage, error)
+	// Census methods
+	CensusPut(idx uint64, req *structs.CensusRequest) error
+	CensusPrune(idx uint64, cutoff time.Time) (int, error)
+	CensusListAll() (uint64, []*state.CensusSnapshot, error)
 }
 
-func NewReportingManager(logger hclog.Logger, deps EntDeps, server ServerDelegate, stateProvider StateDelegate) *ReportingManager {
+func NewReportingManager(logger hclog.Logger, clusterId string, autoReporting bool, deps EntDeps, server ServerDelegate, stateProvider StateDelegate, snapshotRetention time.Duration) *ReportingManager {
+	if snapshotRetention <= 0 {
+		snapshotRetention = DefaultSnapshotRetention
+	}
+
 	rm := &ReportingManager{
-		logger:         logger.Named("reporting"),
-		server:         server,
-		stateProvider:  stateProvider,
-		tickerInterval: ReportingInterval,
+		logger:                 logger.Named("reporting"),
+		clusterId:              clusterId,
+		autoReporting:          autoReporting,
+		server:                 server,
+		stateProvider:          stateProvider,
+		tickerInterval:         ReportingInterval,
+		manualSnapshotInterval: ManualSnapshotInterval,
+		snapshotRetention:      snapshotRetention,
 	}
 	err := rm.initEnterpriseReporting(deps)
 	if err != nil {
 		rm.logger.Error("Error initializing reporting manager", "error", err)
 		return nil
 	}
+	rm.logger.Debug("Created reporting manager")
 	return rm
-}
+}

agent/consul/reporting/reporting_ce.go

@@ -26,6 +26,10 @@ func (rm *ReportingManager) StopReportingAgent() error {
 	return nil
 }
 
-func (m *ReportingManager) Run(ctx context.Context) {
+func (rm *ReportingManager) RunMetricsWriter(ctx context.Context) {
+	// no op
+}
+
+func (rm *ReportingManager) RunManualSnapshotWriter(ctx context.Context) {
 	// no op
 }

agent/consul/server.go

@@ -817,8 +817,9 @@ func NewServer(config *Config, flat Deps, externalGRPCServer *grpc.Server,
 	s.overviewManager = NewOverviewManager(s.logger, s.fsm, s.config.MetricsReportingInterval)
 	go s.overviewManager.Run(&lib.StopChannelContext{StopCh: s.shutdownCh})
 
-	s.reportingManager = reporting.NewReportingManager(s.logger, getEnterpriseReportingDeps(flat), s, s.fsm.State())
-	go s.reportingManager.Run(&lib.StopChannelContext{StopCh: s.shutdownCh})
+	s.reportingManager = reporting.NewReportingManager(s.logger, config.Datacenter, config.Reporting.License.Enabled, getEnterpriseReportingDeps(flat), s, s.fsm.State(), config.Reporting.SnapshotRetentionTime)
+	go s.reportingManager.RunMetricsWriter(&lib.StopChannelContext{StopCh: s.shutdownCh})
+	go s.reportingManager.RunManualSnapshotWriter(&lib.StopChannelContext{StopCh: s.shutdownCh})
 
 	// configure the server specific grpc interfaces (in-process + internal multiplexed grpc)
 	if err := s.setupGRPCInterfaces(config, flat); err != nil {

agent/consul/state/census_utilization.go

@@ -0,0 +1,102 @@
+package state
+
+import (
+	"time"
+
+	"github.com/hashicorp/consul/agent/structs"
+	"github.com/hashicorp/go-memdb"
+)
+
+const censusTable = "census_snapshots"
+
+type CensusSnapshot struct {
+	ID          string
+	TS          time.Time
+	TSUnix      int64
+	Data        []byte
+	CreateIndex uint64
+	ModifyIndex uint64
+}
+
+func censusTableSchema() *memdb.TableSchema {
+	return &memdb.TableSchema{
+		Name: censusTable,
+		Indexes: map[string]*memdb.IndexSchema{
+			"id":     {Name: "id", Unique: true, Indexer: &memdb.StringFieldIndex{Field: "ID"}},
+			"ts":     {Name: "ts", Unique: false, Indexer: &memdb.IntFieldIndex{Field: "TSUnix"}},
+			"create": {Name: "create", Unique: false, Indexer: &memdb.UintFieldIndex{Field: "CreateIndex"}},
+			"modify": {Name: "modify", Unique: false, Indexer: &memdb.UintFieldIndex{Field: "ModifyIndex"}},
+		},
+	}
+}
+
+// Register in the DB schema builder:
+//   Tables[censusTable] = censusTableSchema()
+
+// Write: insert one snapshot at raft index idx.
+func (s *Store) CensusPut(idx uint64, req *structs.CensusRequest) error {
+	tx := s.db.WriteTxn(idx)
+	defer tx.Abort()
+	row := &CensusSnapshot{
+		ID:          req.ID,
+		TS:          req.TS.UTC(),
+		TSUnix:      req.TS.UTC().Unix(),
+		Data:        append([]byte(nil), req.Data...),
+		CreateIndex: idx,
+		ModifyIndex: idx,
+	}
+
+	if err := tx.Insert(censusTable, row); err != nil {
+		return err
+	}
+	if err := tx.Insert(tableIndex, &IndexEntry{censusTable, idx}); err != nil {
+		return err
+	}
+	return tx.Commit()
+}
+
+// Read: list all (already pruned, so this is “everything we keep”).
+func (s *Store) CensusListAll() (uint64, []*CensusSnapshot, error) {
+	txn := s.db.ReadTxn()
+	defer txn.Abort()
+	it, err := txn.Get(censusTable, "ts")
+	if err != nil {
+		return 0, nil, err
+	}
+	var out []*CensusSnapshot
+	for obj := it.Next(); obj != nil; obj = it.Next() {
+		out = append(out, obj.(*CensusSnapshot))
+	}
+	return s.maxIndex(censusTable), out, nil
+}
+
+// Write: prune anything with TS < cutoff. Returns count.
+func (s *Store) CensusPrune(idx uint64, cutoff time.Time) (int, error) {
+	tx := s.db.WriteTxn(idx)
+	defer tx.Abort()
+	cut := cutoff.UTC().Unix()
+
+	it, err := tx.Get(censusTable, "ts")
+	if err != nil {
+		return 0, err
+	}
+
+	c := 0
+	for obj := it.Next(); obj != nil; obj = it.Next() {
+		row := obj.(*CensusSnapshot)
+		if row.TSUnix < cut {
+			if err := tx.Delete(censusTable, row); err != nil {
+				return c, err
+			}
+			c++
+		} else {
+			// rows are ordered by TS; once we reach >= cut we can bail if desired
+			break
+		}
+	}
+
+	if err := tx.Insert(tableIndex, &IndexEntry{censusTable, idx}); err != nil {
+		return c, err
+	}
+	return c, tx.Commit()
+}

agent/consul/state/schema.go

@@ -48,6 +48,7 @@ func newDBSchema() *memdb.DBSchema {
 		tokensTableSchema,
 		tombstonesTableSchema,
 		usageTableSchema,
+		censusTableSchema,
 	)
 	withEnterpriseSchema(db)
 	return db