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

Author: sujay-hashicorp

.changelog/22843.txt

@@ -0,0 +1,5 @@
+```release-note:enhancement
+cmd: Added new subcommand `consul operator utilization [-today-only] [-message] [-y]` to generate a bundle with census utilization snapshot. Main flow is implemented in consul-enterprise
+http: Added a new API Handler for `/v1/operator/utilization`. Core functionality to be implemented in consul-enterprise
+agent: Always enabled census metrics collection with configurable option to export it to Hashicorp Reporting
+```

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/config/testdata/TestRuntimeConfig_Sanitize.golden

@@ -308,7 +308,8 @@
     "Reporting": {
         "License": {
             "Enabled": false
-        }
+        },
+        "SnapshotRetentionTime": "0s"
     },
     "RequestLimitsMode": 0,
     "RequestLimitsReadRate": 0,
@@ -525,4 +526,4 @@
     "Watches": [],
     "XDSUpdateRateLimit": 0,
     "EnableXDSLoadBalancing":true
-}
+}

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
 }