save binding config in database

jobs/metricsforwarder/spec

@@ -24,6 +24,10 @@ templates:
   policy_db.crt.erb: config/certs/policy_db/crt
   policy_db.key.erb: config/certs/policy_db/key
 
+  binding_db_ca.crt.erb: config/certs/binding_db/ca.crt
+  binding_db.crt.erb: config/certs/binding_db/crt
+  binding_db.key.erb: config/certs/binding_db/key
+
   storedprocedure_db_ca.crt.erb: config/certs/storedprocedure_db/ca.crt
   storedprocedure_db.crt.erb: config/certs/storedprocedure_db/crt
   storedprocedure_db.key.erb: config/certs/storedprocedure_db/key

src/autoscaler/api/broker/broker.go

@@ -48,6 +48,8 @@ var (
 	ErrDeleteServiceBinding        = errors.New("error deleting service binding")
 	ErrCredentialNotDeleted        = errors.New("failed to delete custom metrics credential for unbinding")
 	ErrInvalidCredentialType       = errors.New("invalid credential type provided: allowed values are [binding-secret, x509]")
+
+	ErrInvalidConfigurations = errors.New("invalid binding configurations provided")
 )
 
 type Errors []error
@@ -497,6 +499,20 @@ func (b *Broker) Bind(ctx context.Context, instanceID string, bindingID string,
 		policyJson = details.RawParameters
 	}
 
+	// extract custom metrics configs to determine metric submission strategy
+	var bindingConfiguration *models.BindingConfig
+	if policyJson != nil {
+		bindingConfiguration := &models.BindingConfig{}
+		err := json.Unmarshal(policyJson, &bindingConfiguration)
+		if err != nil {
+			actionReadBindingConfiguration := "read-binding-configurations"
+			logger.Error("unmarshal-binding-configuration", err)
+			return result, apiresponses.NewFailureResponseBuilder(
+				ErrInvalidConfigurations, http.StatusBadRequest, actionReadBindingConfiguration).
+				WithErrorKey(actionReadBindingConfiguration).
+				Build()
+		}
+	}
 	policy, err := b.getPolicyFromJsonRawMessage(policyJson, instanceID, details.PlanID)
 	if err != nil {
 		logger.Error("get-default-policy", err)
@@ -529,9 +545,9 @@ func (b *Broker) Bind(ctx context.Context, instanceID string, bindingID string,
 	if err := b.handleExistingBindingsResiliently(ctx, instanceID, appGUID, logger); err != nil {
 		return result, err
 	}
+	// save custom metrics strategy check - bindingConfiguration.CustomMetricsConfig.MetricSubmissionStrategy ! == ""
+	err = createServiceBinding(ctx, b.bindingdb, bindingID, instanceID, appGUID, bindingConfiguration.GetCustomMetricsStrategy())
 
-	// create binding in DB
-	err = b.bindingdb.CreateServiceBinding(ctx, bindingID, instanceID, appGUID)
 	if err != nil {
 		actionCreateServiceBinding := "create-service-binding"
 		logger.Error(actionCreateServiceBinding, err)
@@ -844,3 +860,10 @@ func (b *Broker) deleteBinding(ctx context.Context, bindingId string, serviceIns
 func isValidCredentialType(credentialType string) bool {
 	return credentialType == models.BindingSecret || credentialType == models.X509Certificate
 }
+
+func createServiceBinding(ctx context.Context, bindingDB db.BindingDB, bindingID, instanceID, appGUID string, customMetricsStrategy int) error {
+	if customMetricsStrategy == 1 {
+		return bindingDB.CreateServiceBindingWithConfigs(ctx, bindingID, instanceID, appGUID, customMetricsStrategy)
+	}
+	return bindingDB.CreateServiceBinding(ctx, bindingID, instanceID, appGUID)
+}

src/autoscaler/api/brokerserver/broker_handler_test.go

@@ -907,7 +907,7 @@ var _ = Describe("BrokerHandler", func() {
 				Expect(resp.Body.String()).To(MatchJSON(`{"description":"error: policy did not adhere to plan: Too many scaling rules: Found 2 scaling rules, but a maximum of 1 scaling rules are allowed for this service plan. "}`))
 			})
 		})
-		Context("When mandatory parameters are present", func() {
+		FContext("When mandatory parameters are present", func() {
 			BeforeEach(func() {
 				body, err = json.Marshal(bindingRequestBody)
 				Expect(err).NotTo(HaveOccurred())
@@ -928,7 +928,64 @@ var _ = Describe("BrokerHandler", func() {
 				Expect(creds.Credentials.CustomMetrics.MtlsUrl).To(Equal("Mtls-someURL"))
 			})
 		})
-		// test for credential-type
+		XContext("Binding configurations are present", func() {
+			BeforeEach(func() {
+				bindingPolicy = `{
+				  "configuration": {
+					"custom_metrics": {
+					  "auth": {
+						"credential_type": "binding_secret"
+					  },
+					  "metric_submission_strategy": {
+						"allow_from": "bound_app"
+					  }
+					}
+				  },
+				  "instance_max_count":4,
+				  "instance_min_count":1,
+				  "schedules": {
+					"timezone": "Asia/Shanghai",
+					"recurring_schedule": [{
+					  "start_time": "10:00",
+					  "end_time": "18:00",
+					  "days_of_week": [
+						1,
+						2,
+						3
+					  ],
+					  "instance_min_count": 1,
+					  "instance_max_count": 10,
+					  "initial_min_instance_count": 5
+					}]
+				  },
+				  "scaling_rules":[
+					{
+					  "metric_type":"memoryused",
+					  "threshold":30,
+					  "operator":"<",
+					  "adjustment":"-1"
+					}]
+				}`
+				bindingRequestBody.Policy = json.RawMessage(bindingPolicy)
+				body, err = json.Marshal(bindingRequestBody)
+				Expect(err).NotTo(HaveOccurred())
+			})
+			It("succeeds with 201", func() {
+				Expect(resp.Code).To(Equal(http.StatusCreated))
+
+				By("updating the scheduler")
+				Expect(schedulerServer.ReceivedRequests()).To(HaveLen(1))
+			})
+
+			It("save config to database and returns with 201", func() {
+				// bindingdb.SaveCustomMetricsStrategyReturns(nil)
+				Expect(resp.Code).To(Equal(http.StatusCreated))
+
+				By("CreateServiceBindingWithConfigs should have called one time only")
+				// Expect(bindingdb.SaveCustomMetricsStrategyCallCount()).To(Equal(1))
+			})
+		})
+
 		Context("credential-type is provided while binding", func() {
 			BeforeEach(func() {
 				schedulerExpectedJSON = `{
@@ -1188,7 +1245,6 @@ var _ = Describe("BrokerHandler", func() {
 			})
 		})
 
-		//
 		Context("When a default policy was provided when creating the service instance", func() {
 			BeforeEach(func() {
 				bindingdb.GetServiceInstanceReturns(&models.ServiceInstance{testInstanceId, testOrgId, testSpaceId, testDefaultPolicy, testDefaultGuid}, nil)

src/autoscaler/api/db/servicebroker.db.changelog.yaml

@@ -75,6 +75,11 @@ databaseChangeLog:
                   type: timestamp
                   constraints:
                     nullable: false
+              - column:
+                  name: custom_metrics_strategy
+                  type: int
+                  constraints:
+                    nullable: false
   - changeSet:
       id: 3
       author: qy

src/autoscaler/db/db.go

@@ -80,6 +80,7 @@ type BindingDB interface {
 	GetBindingIdsByInstanceId(ctx context.Context, instanceId string) ([]string, error)
 	GetAppBindingByAppId(ctx context.Context, appId string) (string, error)
 	IsAppBoundToSameAutoscaler(ctx context.Context, appId string, appToScaleId string) (bool, error)
+	CreateServiceBindingWithConfigs(ctx context.Context, bindingId string, serviceInstanceId string, appId string, strategy int) error
 }
 
 type AppMetricDB interface {

src/autoscaler/db/sqldb/binding_sqldb.go

@@ -201,10 +201,28 @@ func (bdb *BindingSQLDB) DeleteServiceInstance(ctx context.Context, serviceInsta
 }
 
 func (bdb *BindingSQLDB) CreateServiceBinding(ctx context.Context, bindingId string, serviceInstanceId string, appId string) error {
+	defaultCustomMetricStrategy := 0 //
+	err := bdb.isBindingExists(ctx, bindingId, serviceInstanceId, appId)
+	if err != nil {
+		return err
+	}
+
+	query := bdb.sqldb.Rebind("INSERT INTO binding" +
+		"(binding_id, service_instance_id, app_id, created_at, custom_metrics_strategy) " +
+		"VALUES(?, ?, ?, ?,?)")
+	_, err = bdb.sqldb.ExecContext(ctx, query, bindingId, serviceInstanceId, appId, time.Now(), defaultCustomMetricStrategy)
+
+	if err != nil {
+		bdb.logger.Error("create-service-binding", err, lager.Data{"query": query, "serviceinstanceid": serviceInstanceId, "bindingid": bindingId, "appid": appId})
+	}
+	return err
+}
+
+func (bdb *BindingSQLDB) isBindingExists(ctx context.Context, bindingId string, serviceInstanceId string, appId string) error {
 	query := bdb.sqldb.Rebind("SELECT * FROM binding WHERE app_id =?")
 	rows, err := bdb.sqldb.QueryContext(ctx, query, appId)
 	if err != nil {
-		bdb.logger.Error("create-service-binding", err, lager.Data{"query": query, "appId": appId, "serviceId": serviceInstanceId, "bindingId": bindingId})
+		bdb.logger.Error("is-binding-already-exists", err, lager.Data{"query": query, "appId": appId, "serviceId": serviceInstanceId, "bindingId": bindingId})
 		return err
 	}
 
@@ -216,19 +234,10 @@ func (bdb *BindingSQLDB) CreateServiceBinding(ctx context.Context, bindingId str
 
 	err = rows.Err()
 	if err != nil {
-		bdb.logger.Error("create-service-binding", err, lager.Data{"query": query, "appId": appId, "serviceId": serviceInstanceId, "bindingId": bindingId})
+		bdb.logger.Error("is-binding-already-exists", err, lager.Data{"query": query, "appId": appId, "serviceId": serviceInstanceId, "bindingId": bindingId})
 		return err
 	}
-
-	query = bdb.sqldb.Rebind("INSERT INTO binding" +
-		"(binding_id, service_instance_id, app_id, created_at) " +
-		"VALUES(?, ?, ?, ?)")
-	_, err = bdb.sqldb.ExecContext(ctx, query, bindingId, serviceInstanceId, appId, time.Now())
-
-	if err != nil {
-		bdb.logger.Error("create-service-binding", err, lager.Data{"query": query, "serviceinstanceid": serviceInstanceId, "bindingid": bindingId, "appid": appId})
-	}
-	return err
+	return nil
 }
 
 func (bdb *BindingSQLDB) GetServiceBinding(ctx context.Context, serviceBindingId string) (*models.ServiceBinding, error) {
@@ -380,11 +389,6 @@ func (bdb *BindingSQLDB) GetBindingIdsByInstanceId(ctx context.Context, instance
 	return bindingIds, rows.Err()
 }
 
-/*
-get all bounded apps to the same autoscaler instance - How to check this? check from the database binding_db database
--> app_id->binding_id->service_instance_id-> all bound apps
-*/
-
 func (bdb *BindingSQLDB) IsAppBoundToSameAutoscaler(ctx context.Context, metricSubmitterAppId string, appToScaleId string) (bool, error) {
 
 	serviceInstanceId, err := bdb.GetServiceInstanceIdByAppId(metricSubmitterAppId)
@@ -404,7 +408,7 @@ func (bdb *BindingSQLDB) IsAppBoundToSameAutoscaler(ctx context.Context, metricS
 	}
 
 	if len(appIds) == 0 {
-		bdb.logger.Error("no-apps-found-by-serviceInstanceId", err, lager.Data{"serviceInstanceId": serviceInstanceId})
+		bdb.logger.Error("no-apps-bounded-with-serviceInstance", err, lager.Data{"serviceInstanceId": serviceInstanceId})
 		return false, nil
 	}
 	// check if the app to scale is in the list of apps bound to the same service instance and return true .otherwise return false
@@ -415,3 +419,19 @@ func (bdb *BindingSQLDB) IsAppBoundToSameAutoscaler(ctx context.Context, metricS
 	}
 	return false, nil
 }
+
+func (bdb *BindingSQLDB) CreateServiceBindingWithConfigs(ctx context.Context, bindingId string, serviceInstanceId string, appId string, strategy int) error {
+	err := bdb.isBindingExists(ctx, bindingId, serviceInstanceId, appId)
+	if err != nil {
+		return err
+	}
+	query := bdb.sqldb.Rebind("INSERT INTO binding" +
+		"(binding_id, service_instance_id, app_id, created_at, custom_metrics_strategy) " +
+		"VALUES(?, ?, ?, ?,?)")
+	_, err = bdb.sqldb.ExecContext(ctx, query, bindingId, serviceInstanceId, appId, time.Now(), strategy)
+
+	if err != nil {
+		bdb.logger.Error("create-service-binding", err, lager.Data{"query": query, "serviceinstanceid": serviceInstanceId, "bindingid": bindingId, "appid": appId, "strategy": strategy})
+	}
+	return err
+}

src/autoscaler/db/sqldb/binding_sqldb_test.go

@@ -632,7 +632,7 @@ var _ = Describe("BindingSqldb", func() {
 			isTestApp1Bounded, _ = bdb.IsAppBoundToSameAutoscaler(context.Background(), testAppId, testAppId2)
 			Expect(err).NotTo(HaveOccurred())
 		})
-		Context("when binding for bindingId exists", func() {
+		When("apps are bounded to same autoscaler instance", func() {
 			BeforeEach(func() {
 				err = bdb.CreateServiceInstance(context.Background(), models.ServiceInstance{ServiceInstanceId: testInstanceId, OrgId: testOrgGuid, SpaceId: testSpaceGuid, DefaultPolicy: policyJsonStr, DefaultPolicyGuid: policyGuid})
 				Expect(err).NotTo(HaveOccurred())
@@ -663,6 +663,34 @@ var _ = Describe("BindingSqldb", func() {
 		})
 
 	})
+
+	Describe("CreateServiceBindingWithConfigs", func() {
+		BeforeEach(func() {
+			err = bdb.CreateServiceInstance(context.Background(), models.ServiceInstance{ServiceInstanceId: testInstanceId, OrgId: testOrgGuid, SpaceId: testSpaceGuid, DefaultPolicy: policyJsonStr, DefaultPolicyGuid: policyGuid})
+			Expect(err).NotTo(HaveOccurred())
+		})
+		Context("When configuration bounded_app is provided", func() {
+			JustBeforeEach(func() {
+				err = bdb.CreateServiceBindingWithConfigs(context.Background(), testBindingId, testInstanceId, testAppId, 1)
+				Expect(err).NotTo(HaveOccurred())
+			})
+			It("should save the binding in the database", func() {
+				Expect(hasServiceBindingWithCustomMetricStrategy(testBindingId, testInstanceId)).To(BeTrue())
+
+			})
+		})
+		Context("When default configuration is provided", func() {
+			JustBeforeEach(func() {
+				err = bdb.CreateServiceBindingWithConfigs(context.Background(), testBindingId, testInstanceId, testAppId, 0)
+				Expect(err).NotTo(HaveOccurred())
+			})
+			It("should save the binding in the database", func() {
+				Expect(hasServiceBindingWithCustomMetricStrategy(testBindingId, testInstanceId)).To(BeFalse())
+
+			})
+		})
+
+	})
 })
 
 func addProcessIdTo(id string) string {

src/autoscaler/db/sqldb/sqldb_suite_test.go

@@ -117,6 +117,16 @@ func hasServiceBinding(bindingId string, serviceInstanceId string) bool {
 	return item
 }
 
+func hasServiceBindingWithCustomMetricStrategy(bindingId string, serviceInstanceId string) bool {
+	query := dbHelper.Rebind("SELECT * FROM binding WHERE binding_id = ? AND service_instance_id = ? AND custom_metrics_strategy = 1")
+	rows, e := dbHelper.Query(query, bindingId, serviceInstanceId)
+	FailOnError("can not query table binding", e)
+	defer func() { _ = rows.Close() }()
+	item := rows.Next()
+	FailOnError("can not query table binding", rows.Err())
+	return item
+}
+
 func cleanPolicyTable() {
 	_, e := dbHelper.Exec("DELETE from policy_json")
 	if e != nil {

src/autoscaler/eventgenerator/metric/fetcher_test.go

@@ -300,7 +300,7 @@ var _ = Describe("logCacheFetcher", func() {
 				Entry("metric type cpuutil", models.MetricNameCPUUtil, "cpu_entitlement"),
 				Entry("metric type disk", models.MetricNameDisk, "disk"),
 				Entry("metric type diskutil", models.MetricNameDiskUtil, "disk|disk_quota"),
-				Entry("metric type CustomMetrics", "a-custom-metric", "a-custom-metric"),
+				Entry("metric type CustomMetricsConfig", "a-custom-metric", "a-custom-metric"),
 			)
 		})
 	})

src/autoscaler/metricsforwarder/server/auth/custom_metrics_strategy.go

@@ -0,0 +1,59 @@
+package auth
+
+import (
+	"code.cloudfoundry.org/app-autoscaler/src/autoscaler/db"
+	"code.cloudfoundry.org/lager/v3"
+	"net/http"
+	"strings"
+)
+
+type MetricsSubmissionStrategy interface {
+	validate(appId string, submitterAppIdFromCert string, logger lager.Logger, bindingDB db.BindingDB, r *http.Request) error
+}
+
+var _ MetricsSubmissionStrategy = &DefaultMetricsSubmissionStrategy{}
+
+type DefaultMetricsSubmissionStrategy struct{}
+
+func (d *DefaultMetricsSubmissionStrategy) validate(appId string, submitterAppIdFromCert string, logger lager.Logger, bindingDB db.BindingDB, r *http.Request) error {
+	// check if appID is same as AppIdFromCert
+	if appId != submitterAppIdFromCert {
+		return ErrorAppIDWrong
+	}
+	return nil
+}
+
+type BoundedMetricsSubmissionStrategy struct{}
+
+func (c *BoundedMetricsSubmissionStrategy) validate(appId string, submitterAppIdFromCert string, logger lager.Logger, bindingDB db.BindingDB, r *http.Request) error {
+	// check if appID is same as AppIdFromCert
+	if appId != submitterAppIdFromCert {
+		c.verifyMetricSubmissionStrategy(r, logger, bindingDB, submitterAppIdFromCert, appId)
+	}
+	return nil
+}
+
+func (c *BoundedMetricsSubmissionStrategy) verifyMetricSubmissionStrategy(r *http.Request, logger lager.Logger, bindingDB db.BindingDB, submitterAppCert string, appID string) (bool, error) {
+
+	customMetricSubmissionStrategy := r.Header.Get("custom-metrics-submission-strategy")
+	customMetricSubmissionStrategy = strings.ToLower(customMetricSubmissionStrategy)
+	if customMetricSubmissionStrategy == "" {
+		logger.Info("custom-metrics-submission-strategy-not-found", lager.Data{"appID": appID})
+		return false, nil
+	}
+	if customMetricSubmissionStrategy == "bound_app" {
+		logger.Info("custom-metrics-submission-strategy-found", lager.Data{"appID": appID, "strategy": customMetricSubmissionStrategy})
+		// check if the app is bound to same autoscaler instance by check the binding id from the bindingdb
+		// if the app is bound to the same autoscaler instance, then allow the request to the next handler i.e publish custom metrics
+		isAppBound, err := bindingDB.IsAppBoundToSameAutoscaler(r.Context(), submitterAppCert, appID)
+		if err != nil {
+			logger.Error("error-checking-app-bound-to-same-service", err, lager.Data{"metric-submitter-app-id": submitterAppCert})
+			return false, err
+		}
+		if isAppBound == false {
+			logger.Info("app-not-bound-to-same-service", lager.Data{"app-id": submitterAppCert})
+			return false, ErrorAppNotBound
+		}
+	}
+	return true, nil
+}