Making Enhanced Compliance Reporting configurable (#7477)

* Making Enhanced Compliance Reporting configurable

Signed-off-by: Durga Sarat Chandra Maddu <dmaddu@progress.com>

* Adding delta upgrade logic

Signed-off-by: Durga Sarat Chandra Maddu <dmaddu@progress.com>

* Serving Enhanced Reporting APIs based on configuration

Signed-off-by: Durga Sarat Chandra Maddu <dmaddu@progress.com>

* updating getStartDateFromEndDate based on enhanced_reporting configuration

Signed-off-by: Durga Sarat Chandra Maddu <dmaddu@progress.com>

* adding upgrade time for upgrade process (#7490)

* adding upgrade time for upgrade process

Signed-off-by: Yashvi Jain <yashvi.jain@progress.com>

* adding the correct names

Signed-off-by: Yashvi Jain <yashvi.jain@progress.com>

Signed-off-by: Yashvi Jain <yashvi.jain@progress.com>

* control report with date range logic update (#7491)

* logic fix

Signed-off-by: Abdul-Az <aazeez@progress.com>

* modifying the control summary

Signed-off-by: Durga Sarat Chandra Maddu <dmaddu@progress.com>

* error message fix

Signed-off-by: Abdul-Az <aazeez@progress.com>

Signed-off-by: Abdul-Az <aazeez@progress.com>
Signed-off-by: Durga Sarat Chandra Maddu <dmaddu@progress.com>
Co-authored-by: dmaddu <dmaddu@progress.com>

* Addressing review comments

Signed-off-by: Durga Sarat Chandra Maddu <dmaddu@progress.com>

Signed-off-by: Durga Sarat Chandra Maddu <dmaddu@progress.com>
Signed-off-by: Yashvi Jain <yashvi.jain@progress.com>
Signed-off-by: Abdul-Az <aazeez@progress.com>
Co-authored-by: dmaddu <dmaddu@progress.com>
Co-authored-by: Yashvi Jain <91940132+YashviJain01@users.noreply.github.com>
Co-authored-by: Abdul Azeez <aazeez@progress.com>

api/config/compliance/config_request.proto

@@ -45,6 +45,7 @@ message ConfigRequest {
 				google.protobuf.Int32Value message_buffer_size = 8;
 				google.protobuf.BoolValue enable_large_reporting = 9;
 				google.protobuf.Int32Value lcr_open_search_requests = 10;
+				google.protobuf.BoolValue enable_enhanced_compliance_reporting = 11;
 			}
 
 			message Proxy {

components/compliance-service/api/reporting/server/server.go

@@ -34,18 +34,21 @@ const streamBufferSize = 2097152
 
 // Server implementation for reporting
 type Server struct {
-	es                       *relaxting.ES2Backend
-	reportMgr                reportmanager.ReportManagerServiceClient
-	lcr_open_search_requests int
-	db                       *pgdb.DB
+	es                         *relaxting.ES2Backend
+	reportMgr                  reportmanager.ReportManagerServiceClient
+	lcr_open_search_requests   int
+	db                         *pgdb.DB
+	isEnhancedReportingEnabled bool
 }
 
 // New creates a new server
-func New(es *relaxting.ES2Backend, rm reportmanager.ReportManagerServiceClient, lcrOpenSearchRequest int, db *pgdb.DB) *Server {
+func New(es *relaxting.ES2Backend, rm reportmanager.ReportManagerServiceClient, lcrOpenSearchRequest int, db *pgdb.DB,
+	isEnhancedReportingEnabled bool) *Server {
 	server := Server{
-		es:                       es,
-		lcr_open_search_requests: lcrOpenSearchRequest,
-		db:                       db,
+		es:                         es,
+		lcr_open_search_requests:   lcrOpenSearchRequest,
+		db:                         db,
+		isEnhancedReportingEnabled: isEnhancedReportingEnabled,
 	}
 	if rm != nil {
 		server.reportMgr = rm
@@ -917,7 +920,11 @@ func (srv *Server) GetReportContent(ctx context.Context, in *reporting.ReportCon
 	}
 	return nil*/
 }
+
 func (srv *Server) AssetCount(ctx context.Context, in *reporting.ListFilters) (*reporting.AssetSummary, error) {
+	if !srv.isEnhancedReportingEnabled {
+		return nil, status.Error(codes.PermissionDenied, "customer not enabled for enhanced compliance reporting")
+	}
 	var assets *reporting.AssetSummary
 
 	formattedFilters := formatFilters(in.Filters)
@@ -938,6 +945,9 @@ func (srv *Server) AssetCount(ctx context.Context, in *reporting.ListFilters) (*
 }
 
 func (srv *Server) ListAsset(ctx context.Context, in *reporting.AssetListRequest) (*reporting.AssetListResponse, error) {
+	if !srv.isEnhancedReportingEnabled {
+		return nil, status.Error(codes.PermissionDenied, "customer not enabled for enhanced compliance reporting")
+	}
 	formattedFilters := formatFilters(in.Filters)
 	var asset []*reporting.Assets
 
@@ -955,6 +965,9 @@ func (srv *Server) ListAsset(ctx context.Context, in *reporting.AssetListRequest
 }
 
 func (srv *Server) GetAssetConfig(ctx context.Context, in *reporting.GetAssetConfigRequest) (*reporting.ComplianceConfigResponse, error) {
+	if !srv.isEnhancedReportingEnabled {
+		return nil, status.Error(codes.PermissionDenied, "customer not enabled for enhanced compliance reporting")
+	}
 	result, err := srv.db.GetConfigs(ctx)
 	if err != nil {
 		logrus.Errorf("error while getting the conf: %+v", err)
@@ -965,6 +978,9 @@ func (srv *Server) GetAssetConfig(ctx context.Context, in *reporting.GetAssetCon
 }
 
 func (srv *Server) SetAssetConfig(ctx context.Context, in *reporting.ComplianceConfigRequest) (*reporting.ComplianceConfigResponse, error) {
+	if !srv.isEnhancedReportingEnabled {
+		return nil, status.Error(codes.PermissionDenied, "customer not enabled for enhanced compliance reporting")
+	}
 	err := srv.db.SetConfigs(ctx, in)
 	if err != nil {
 		logrus.Errorf("error while updating the conf: %+v", err)
@@ -982,6 +998,9 @@ func (srv *Server) SetAssetConfig(ctx context.Context, in *reporting.ComplianceC
 
 // ListControlItemsRange returns a list of controlListItems based on query
 func (srv *Server) ListControlItemsRange(ctx context.Context, in *reporting.ControlItemRequest) (*reporting.ControlItems, error) {
+	if !srv.isEnhancedReportingEnabled {
+		return nil, status.Error(codes.PermissionDenied, "customer not enabled for enhanced compliance reporting")
+	}
 	var controlListItems *reporting.ControlItems
 	if in.Size == 0 {
 		in.Size = 100

components/compliance-service/cmd/compliance-service/cmd/run.go

@@ -109,6 +109,7 @@ func init() {
 	runCmd.Flags().IntVar(&conf.Service.MessageBufferSize, "message-buffer-size", 100, "Number of ingest messages allowed to buffer")
 	runCmd.Flags().BoolVar(&conf.Service.EnableLargeReporting, "enable-large-reporting", false, "upgrade to support large reporting")
 	runCmd.Flags().IntVar(&conf.Service.LcrOpenSearchRequests, "lcr-open-search-requests", conf.Service.LcrOpenSearchRequests, "number of concurrent requests to communicate with open search for large compliance reporting")
+	runCmd.Flags().BoolVar(&conf.Service.EnableEnhancedReporting, "enable-enhanced-reporting", false, "upgrade to support enhanced compliance reporting")
 
 	// Postgres Config Flags
 	runCmd.Flags().StringVar(&conf.Postgres.ConnectionString, "postgres-uri", conf.Postgres.ConnectionString, "PostgreSQL connection string to use")

components/compliance-service/compliance.go

@@ -85,11 +85,12 @@ func createESBackend(servConf *config.Compliance, db *pgdb.DB) relaxting.ES2Back
 
 	// define the ElasticSearch backend config with legacy automate auth
 	esr := relaxting.ES2Backend{
-		ESUrl:             servConf.ElasticSearch.Url,
-		Enterprise:        servConf.Delivery.Enterprise,
-		ChefDeliveryUser:  servConf.Delivery.User,
-		ChefDeliveryToken: servConf.Delivery.Token,
-		PGdb:              db,
+		ESUrl:                      servConf.ElasticSearch.Url,
+		Enterprise:                 servConf.Delivery.Enterprise,
+		ChefDeliveryUser:           servConf.Delivery.User,
+		ChefDeliveryToken:          servConf.Delivery.Token,
+		PGdb:                       db,
+		IsEnhancedReportingEnabled: servConf.Service.EnableEnhancedReporting,
 	}
 	return esr
 }
@@ -176,7 +177,7 @@ func serveGrpc(ctx context.Context, db *pgdb.DB, connFactory *secureconn.Factory
 		logrus.Infof("not getting authz client; env var RUN_MODE found. value is 'test' ")
 	}
 	nodeManagerServiceClient := getManagerConnection(connFactory, conf.Manager.Endpoint)
-	ingesticESClient := ingestic.NewESClient(esClient)
+	ingesticESClient := ingestic.NewESClient(esClient, conf)
 	ingesticESClient.InitializeStore(context.Background())
 	runner.ESClient = ingesticESClient
 	var reportmanagerClient reportmanager.ReportManagerServiceClient
@@ -209,11 +210,12 @@ func serveGrpc(ctx context.Context, db *pgdb.DB, connFactory *secureconn.Factory
 
 	upgradeDB := pgdb.NewDB(db)
 	upgradeService := migrations.NewService(upgradeDB, cerealManager)
-
-	// Initiating cereal Manager for upgrade jobs
-	err = migrations.InitCerealManager(cerealManager, 1, ingesticESClient, upgradeDB)
-	if err != nil {
-		logrus.Fatalf("Failed to initiate cereal manager for upgrading jobs %v", err)
+	if conf.Service.EnableEnhancedReporting {
+		// Initiating cereal Manager for upgrade jobs
+		err = migrations.InitCerealManager(cerealManager, 1, ingesticESClient, upgradeDB)
+		if err != nil {
+			logrus.Fatalf("Failed to initiate cereal manager for upgrading jobs %v", err)
+		}
 	}
 
 	err = processor.InitCerealManager(cerealManager, conf.CerealConfig.Workers, ingesticESClient)
@@ -230,7 +232,7 @@ func serveGrpc(ctx context.Context, db *pgdb.DB, connFactory *secureconn.Factory
 	jobs.RegisterJobsServiceServer(s, jobsserver.New(db, connFactory, eventClient,
 		conf.Manager.Endpoint, cerealManager))
 	reporting.RegisterReportingServiceServer(s, reportingserver.New(&esr, reportmanagerClient,
-		conf.Service.LcrOpenSearchRequests, db))
+		conf.Service.LcrOpenSearchRequests, db, conf.Service.EnableEnhancedReporting))
 
 	ps := profilesserver.New(db, &esr, ingesticESClient, &conf.Profiles, eventClient, statusSrv)
 	profiles.RegisterProfilesServiceServer(s, ps)
@@ -272,8 +274,14 @@ func serveGrpc(ctx context.Context, db *pgdb.DB, connFactory *secureconn.Factory
 		logrus.Fatalf("serveGrpc aborting, unable to run migrations: %v", err)
 	}
 
-	// Running upgrade scenarios for DayLatest flag
-	go upgradeService.PollForUpgradeFlagDayLatest()
+	date, err := upgradeService.UpdateFlags(conf.Service.EnableEnhancedReporting)
+	if err != nil {
+		logrus.Fatalf("serveGrpc aborting, unable to find the date to run the upgrades: %v", err)
+	}
+	if conf.Service.EnableEnhancedReporting {
+		// Running upgrade scenarios for DayLatest flag
+		go upgradeService.PollForUpgradeFlagDayLatest(date)
+	}
 
 	errc := make(chan error)
 	defer close(errc)

components/compliance-service/config/types.go

@@ -13,10 +13,11 @@ type Service struct {
 	LogLevel   string
 	ServerBind string
 	certs.TLSConfig
-	ConfigFilePath        string
-	MessageBufferSize     int
-	EnableLargeReporting  bool
-	LcrOpenSearchRequests int
+	ConfigFilePath          string
+	MessageBufferSize       int
+	EnableLargeReporting    bool
+	LcrOpenSearchRequests   int
+	EnableEnhancedReporting bool
 }
 
 // Compliance service specific config options

components/compliance-service/dao/pgdb/migration/sql/44_alter_upgrade_time_on_upgrade_flags.down.sql

@@ -0,0 +1,2 @@
+ALTER TABLE IF EXISTS upgrade_flags
+  DROP COLUMN IF EXISTS upgrade_time;

components/compliance-service/dao/pgdb/migration/sql/44_alter_upgrade_time_on_upgrade_flags.up.sql

@@ -0,0 +1,2 @@
+ALTER TABLE IF EXISTS upgrade_flags
+  ADD COLUMN IF NOT EXISTS upgrade_time TIMESTAMP DEFAULT '0001-01-01T00:00:00Z00:00';

components/compliance-service/dao/pgdb/migration/sql/45_add_enhanced_reporting.up.sql

@@ -0,0 +1,26 @@
+INSERT INTO UPGRADE_FLAGS(
+  id, upgrade_flag, upgrade_value, upgrade_time
+) 
+select 
+  * 
+from 
+  (
+    select 
+      3 as id, 
+      'enhanced_reporting' as upgrade_flag, 
+      false as upgrade_value, 
+      TO_TIMESTAMP(
+        '2022-07-18 01:00:00', 'YYYY-MM-DD HH:MI:SS'
+      ) as upgrade_time
+  ) as tmp 
+where 
+  not exists (
+    select 
+      upgrade_flag 
+    from 
+      UPGRADE_FLAGS 
+    where 
+      upgrade_flag = 'enhanced_reporting' 
+    limit 
+      1
+  );

components/compliance-service/dao/pgdb/storage.go

@@ -1,17 +1,25 @@
 package pgdb
 
+import "time"
+
 type Storage interface {
-	GetUpgradeFlags() (map[string]bool, error)
-	UpdateControlFlagToFalse() error
+	GetUpgradeFlags() (map[string]Flag, error)
+	UpdateControlFlagValue(bool) error
+	UpdateControlFlagTimeStamp() error
+	AddEnhancedReportingFlag() error
+	RemoveEnhancedReportingFlag() error
 }
 
-const DayLatestFlag = "day_latest"
-
-const ControlIndexFlag = "control_index"
-
-const CompRunInfoFlag = "comp_run_info"
+// constants represents the flags
+const (
+	DayLatestFlag                = "day_latest"
+	ControlIndexFlag             = "control_index"
+	CompRunInfoFlag              = "comp_run_info"
+	EnhancedReportingEnabledFlag = "enhanced_reporting"
+)
 
 type Flag struct {
-	flag   string
-	status bool
+	FlagName     string    `db:"upgrade_flag"`
+	Status       bool      `db:"upgrade_value"`
+	UpgradedTime time.Time `db:"upgrade_time"`
 }

components/compliance-service/dao/pgdb/upgrades.go

@@ -2,9 +2,11 @@ package pgdb
 
 import (
 	"fmt"
+	"strings"
+	"time"
+
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
-	"strings"
 )
 
 type UpgradesDB struct {
@@ -15,21 +17,21 @@ func NewDB(db *DB) *UpgradesDB {
 	return &UpgradesDB{db}
 }
 
-//UpdateControlFlagToFalse updates the control index flags to false
-func (u *UpgradesDB) UpdateControlFlagToFalse() error {
-	_, err := u.DB.Exec(getUpdateQuery(ControlIndexFlag))
+//UpdateControlFlagValue updates the upgrade_value for control index flag
+func (u *UpgradesDB) UpdateControlFlagValue(value bool) error {
+	_, err := u.DB.Exec(getUpdateQueryForValue(), value, ControlIndexFlag)
 	if err != nil {
-		return errors.Wrapf(err, "Unable to Control Index flag to db")
+		return errors.Wrapf(err, "Unable to set Control Index upgrade_value to %t", value)
 	}
 	return nil
 }
 
 //GetUpgradeFlags Gets the all the upgrade flags and status from the pg database
-func (u *UpgradesDB) GetUpgradeFlags() (map[string]bool, error) {
-	flagMap := make(map[string]bool)
+func (u *UpgradesDB) GetUpgradeFlags() (map[string]Flag, error) {
+	flagMap := make(map[string]Flag)
 
 	logrus.Info("Inside the comp run info flag")
-	flags := []string{ControlIndexFlag}
+	flags := []string{ControlIndexFlag, EnhancedReportingEnabledFlag}
 	rows, err := u.DB.Query(getQueryForFlag(flags))
 	if err != nil {
 		return flagMap, err
@@ -42,25 +44,64 @@ func (u *UpgradesDB) GetUpgradeFlags() (map[string]bool, error) {
 
 	for rows.Next() {
 		flag := Flag{}
-		if err := rows.Scan(&flag.flag, &flag.status); err != nil {
+		if err := rows.Scan(&flag.FlagName, &flag.Status, &flag.UpgradedTime); err != nil {
 			logrus.Errorf("Unable to get the flags with error %v", err)
 			return nil, err
 		}
-		flagMap[flag.flag] = flag.status
+		flagMap[flag.FlagName] = flag
 	}
 	if err := rows.Err(); err != nil {
 		return nil, errors.Wrap(err, "error retrieving result rows")
 	}
 	return flagMap, err
 }
 
+// UpdateControlFlagTimeStamp updates the upgrade_time for the control index flag
+func (u *UpgradesDB) UpdateControlFlagTimeStamp() error {
+	_, err := u.DB.Exec(getUpdateQueryForTime(ControlIndexFlag), time.Now(), ControlIndexFlag)
+	if err != nil {
+		err = errors.Wrapf(err, "Unable to update the upgrade_time of upgrade_flags for control_index flag")
+	}
+	return err
+}
+
+// AddEnhancedReportingFlag adds the enhanced_reporting flag to flags table
+func (u *UpgradesDB) AddEnhancedReportingFlag() error {
+	_, err := u.DB.Exec(insertQuery(), 3, EnhancedReportingEnabledFlag, false, time.Now())
+	if err != nil {
+		err = errors.Wrapf(err, "Unable to add the enhanced_reporting flag to upgrade_flags table")
+	}
+	return err
+}
+
+// RemoveEnhancedReportingFlag delete the enhanced_reporting from flags table
+func (u *UpgradesDB) RemoveEnhancedReportingFlag() error {
+	_, err := u.DB.Exec(deleteFlag(), EnhancedReportingEnabledFlag)
+	if err != nil {
+		err = errors.Wrapf(err, "Unable to remove the enhanced_reporting flag from upgrade_flags")
+	}
+	return err
+}
+
 //getQueryForFlag gets the query for flag
 func getQueryForFlag(flag []string) string {
 	flags := `'` + strings.Join(flag, `','`) + `'`
-	return fmt.Sprintf("Select upgrade_flag,upgrade_value from upgrade_flags where upgrade_flag in (%s)", flags)
+	return fmt.Sprintf("Select upgrade_flag,upgrade_value, upgrade_time from upgrade_flags where upgrade_flag in (%s)", flags)
+}
+
+//getUpdateQueryForValue gets the update query for setting the upgrade_value
+func getUpdateQueryForValue() string {
+	return fmt.Sprintf("Update upgrade_flags set upgrade_value= $1 where upgrade_flag= $2")
+}
+
+func getUpdateQueryForTime(flag string) string {
+	return fmt.Sprintf("Update upgrade_flags set upgrade_time= $1 where upgrade_flag= $2")
+}
+
+func insertQuery() string {
+	return fmt.Sprintf("INSERT INTO UPGRADE_FLAGS (id,upgrade_flag,upgrade_value,upgrade_time) VALUES ($1,$2,$3,$4)")
 }
 
-//getUpdateQuery gets the update query for flag
-func getUpdateQuery(flag string) string {
-	return fmt.Sprintf("Update upgrade_flags set upgrade_value=false where upgrade_flag='%s'", flag)
+func deleteFlag() string {
+	return fmt.Sprintf("delete from upgrade_flags where upgrade_flag = $1")
 }

components/compliance-service/habitat/default.toml

@@ -11,6 +11,7 @@ port = 10121
 external_fqdn = "localhost"
 enable_large_reporting = false
 lcr_open_search_requests = 50
+enable_enhanced_compliance_reporting = false
 
 [storage]
 database = "chef_compliance_service"

components/compliance-service/habitat/hooks/run

@@ -58,6 +58,7 @@ CONFIG="$CONFIG --log-level {{cfg.logger.level}}"
 CONFIG="$CONFIG --port {{cfg.service.port}}"
 CONFIG="$CONFIG --enable-large-reporting={{cfg.service.enable_large_reporting}}"
 CONFIG="$CONFIG --lcr-open-search-requests {{cfg.service.lcr_open_search_requests}}"
+CONFIG="$CONFIG --enable-enhanced-reporting={{cfg.service.enable_enhanced_compliance_reporting}}"
 
 # Interval in minutes to poll for node status.
 CONFIG="$CONFIG --manager-awsec2-poll {{cfg.nodemanager.awsec2_polling_interval}}"