enable memoryutil based scaling (cloudfoundry#53)

* enable memoryutil based scaling
- bump latest app-autoscaler code
- bosh release code change accordingly
- add acceptance test for memoryutil based scaling
- refactor other acceptance tests

* sync package specs

example/property-overrides.yml

@@ -20,6 +20,10 @@ api_server_properties:
     ca_cert: (( app_autoscaler_ca_cert ))
     client_cert: (( scheduler_client_cert ))
     client_key: (( scheduler_client_key ))
+  scaling_engine:
+    ca_cert: (( app_autoscaler_ca_cert ))
+    client_cert: (( scalingengine_client_cert ))
+    client_key: (( scalingengine_client_key ))
 
 service_broker_properties:
   db_config:

jobs/apiserver/spec

@@ -9,6 +9,9 @@ templates:
   apiserver_ca.crt.erb: config/certs/apiserver/ca.crt
   apiserver_server.crt.erb: config/certs/apiserver/server.crt
   apiserver_server.key.erb: config/certs/apiserver/server.key
+  scalingengine_ca.crt.erb: config/certs/scalingengine/ca.crt
+  scalingengine_client.crt.erb: config/certs/scalingengine/client.crt
+  scalingengine_client.key.erb: config/certs/scalingengine/client.key
   scheduler_ca.crt.erb: config/certs/scheduler/ca.crt
   scheduler_client.crt.erb: config/certs/scheduler/client.crt
   scheduler_client.key.erb: config/certs/scheduler/client.key
@@ -30,6 +33,7 @@ properties:
     description: "PEM-encoded server certificate"
   api_server.server_key:
     description: "PEM-encoded server key"
+
   api_server.scheduler.host:
     description: "Host where scheduler is running"
     default: "scheduler.service.cf.internal"
@@ -42,6 +46,23 @@ properties:
     description: "PEM-encoded client certificate"
   api_server.scheduler.client_key:
     description: "PEM-encoded client key"
+
+
+  api_server.scaling_engine.host:
+    description: "Host where scalingengine is running"
+    default: "scalingengine.service.cf.internal"
+  api_server.scaling_engine.port:
+    description: "Port where scalingengine will listen"
+    default: 6104
+  api_server.scaling_engine.ca_cert:
+    description: "PEM-encoded CA certificate"
+  api_server.scaling_engine.client_cert:
+    description: "PEM-encoded client certificate"
+  api_server.scaling_engine.client_key:
+    description: "PEM-encoded client key"
+
+
+
   api_server.db_config.max_connections:
     description: "Maximum number of connections that may be held in the apiserver db pool"
   api_server.db_config.min_connections:

jobs/apiserver/templates/config.json.erb

@@ -11,6 +11,9 @@
   database = p_arr('policy_db.databases').find { |database| database['tag'] == 'policydb' or database['tag'] == 'default'} 
   scheduler_host = p('api_server.scheduler.host')
   scheduler_port = p('api_server.scheduler.port')
+  scaling_engine_host = p('api_server.scaling_engine.host')
+  scaling_engine_port = p('api_server.scaling_engine.port')
+
   params = {
     'port' => p('api_server.port'),
     'db' => {
@@ -27,6 +30,14 @@
            'caCertFile' => "/var/vcap/jobs/apiserver/config/certs/scheduler/ca.crt"
        },
     },
+    'scalingEngine' => {
+       'uri' => "https://" + scaling_engine_host + ":" + scaling_engine_port.to_s,
+       'tls' => {
+           'keyFile' => "/var/vcap/jobs/apiserver/config/certs/scalingengine/client.key",
+           'certFile' => "/var/vcap/jobs/apiserver/config/certs/scalingengine/client.crt",
+           'caCertFile' => "/var/vcap/jobs/apiserver/config/certs/scalingengine/ca.crt"
+       },
+    },
     'tls' => {
         'keyFile' => "/var/vcap/jobs/apiserver/config/certs/apiserver/server.key",
         'certFile' => "/var/vcap/jobs/apiserver/config/certs/apiserver/server.crt",

jobs/apiserver/templates/scalingengine_ca.crt.erb

@@ -0,0 +1,3 @@
+<% if_p("api_server.scaling_engine.ca_cert") do |value| %>
+<%= value %>
+<% end %>

jobs/apiserver/templates/scalingengine_client.crt.erb

@@ -0,0 +1,3 @@
+<% if_p("api_server.scaling_engine.client_cert") do |value| %>
+<%= value %>
+<% end %>

jobs/apiserver/templates/scalingengine_client.key.erb

@@ -0,0 +1,3 @@
+<% if_p("api_server.scaling_engine.client_key") do |value| %>
+<%= value %>
+<% end %>

jobs/metricscollector/spec

@@ -69,8 +69,12 @@ properties:
     description: "the time interval to refresh app policies from policy database"
     default: 60s
   metricscollector.collector.poll_interval:
-    description: "the time interval to poll container metrics from loggregator"
+    description: "the time interval to collect container metrics from loggregator"
     default: 30s
+  metricscollector.collector.collect_method:
+    description: "the method to collect metrics from loggregator"
+    default: "polling"
+
 
   metricscollector.lock.lock_ttl:
     description: "consul lock ttl duration"

jobs/metricscollector/templates/metricscollector.yml.erb

@@ -45,6 +45,7 @@ db:
 collector:
   refresh_interval: <%= p("metricscollector.collector.refresh_interval") %>
   poll_interval: <%= p("metricscollector.collector.poll_interval") %>
+  collect_method: <%= p("metricscollector.collector.collect_method") %>
 
 lock:
   lock_ttl: <%= p("metricscollector.lock.lock_ttl") %>

jobs/scheduler/spec

@@ -35,6 +35,7 @@ properties:
     description: "Maximum no of jobs can be re-scheduled"
   scheduler.notification_reschedule_maxcount:
     description: "Maximum number of notification sent to scaling engine for job re-schedule"
+
   scheduler.scaling_engine.host:
     description: "URL where Scaling-engine is running"
     default: "scalingengine.service.cf.internal"
@@ -47,9 +48,11 @@ properties:
     description: "PEM-encoded client certificate"
   scheduler.scaling_engine.client_key:
     description: "PEM-encoded client key"
+
   scheduler.consul.ttl:
     description: "TTL timeout seconds"
     default: 20
+
   scheduler_db.address:
     description: "IP address on which the schedulerdb server will listen"
     default: "postgres.service.cf.internal"
@@ -60,4 +63,16 @@ properties:
   scheduler_db.port:
     description: "Port on which the schedulerdb server will listen"
   scheduler_db.roles:
-    description: "The list of database roles used in schedulerdb database including name/password"
+    description: "The list of database roles used in schedulerdb database including name/password"
+
+  policy_db.address:
+    description: "IP address on which the policydb server will listen"
+    default: "postgres.service.cf.internal"
+  policy_db.databases:
+    description: "The list of databases used in policydb database including name"
+  policy_db.db_scheme:
+    description: "Database scheme to be used to access policydb"
+  policy_db.port:
+    description: "Port on which the policydb server will listen"
+  policy_db.roles:
+    description: "The list of database roles used in policydb database including name/password"  

jobs/scheduler/templates/application.properties.erb

@@ -6,15 +6,27 @@
     p(property,nil) || []
   end
 
-  role = p_arr('scheduler_db.roles').find { |role| role['tag'] == 'schedulerdb' or role['tag'] == 'default' } 
-  database = p_arr('scheduler_db.databases').find { |database| database['tag'] == 'schedulerdb' or database['tag'] == 'default' } 
+  schedulerdb_role = p_arr('scheduler_db.roles').find { |role| role['tag'] == 'schedulerdb' or role['tag'] == 'default' } 
+  schedulerdb_database = p_arr('scheduler_db.databases').find { |database| database['tag'] == 'schedulerdb' or database['tag'] == 'default' } 
+
+  policydb_role = p_arr('policy_db.roles').find { |role| role['tag'] == 'policydb' or role['tag'] == 'default' } 
+  policydb_database = p_arr('policy_db.databases').find { |database| database['tag'] == 'policydb' or database['tag'] == 'default' } 
+
 %>
 #datasource for application and quartz
 
 spring.datasource.driverClassName=org.postgresql.Driver
-spring.datasource.url=jdbc:postgresql://<%=p('scheduler_db.address')%>:<%=p('scheduler_db.port') %>/<%=database['name'] %>
-spring.datasource.username=<%=role['name'] %>
-spring.datasource.password=<%=role['password'] %>
+spring.datasource.url=jdbc:postgresql://<%=p('scheduler_db.address')%>:<%=p('scheduler_db.port') %>/<%=schedulerdb_database['name'] %>
+spring.datasource.username=<%=schedulerdb_role['name'] %>
+spring.datasource.password=<%=schedulerdb_role['password'] %>
+
+#datasource for policy 
+
+spring.policyDbDataSource.driverClassName=org.postgresql.Driver
+spring.policyDbDataSource.url=jdbc:postgresql://<%=p('policy_db.address')%>:<%=p('policy_db.port') %>/<%=policydb_database['name'] %>
+spring.policyDbDataSource.password=<%=policydb_role['name'] %>
+spring.policyDbDataSource.username=<%=policydb_role['password'] %>
+
 
 #quartz job
 scalingenginejob.reschedule.interval.millisecond=<%=p('scheduler.job_reschedule_interval_millisecond') %>

packages/metricscollector/spec

@@ -16,6 +16,7 @@ files:
 - app-autoscaler/src/autoscaler/metricscollector/fakes/*.go # gosub
 - app-autoscaler/src/autoscaler/metricscollector/noaa/*.go # gosub
 - app-autoscaler/src/autoscaler/metricscollector/server/*.go # gosub
+- app-autoscaler/src/autoscaler/metricscollector/testhelpers/*.go # gosub
 - app-autoscaler/src/autoscaler/models/*.go # gosub
 - app-autoscaler/src/autoscaler/routes/*.go # gosub
 - app-autoscaler/src/code.cloudfoundry.org/cfhttp/*.go # gosub

src/acceptance/app/app_suite_test.go

@@ -205,9 +205,9 @@ func averageMemoryUsedByInstance(appGUID string, timeout time.Duration) uint64 {
 	return memSum / uint64(len(memoryUsedArray))
 }
 
-func generateDynamicScaleOutPolicy(instanceMin, instanceMax int, threshold int64) string {
+func generateDynamicScaleOutPolicy(instanceMin, instanceMax int, metricName string, threshold int64) string {
 	scalingOutRule := ScalingRule{
-		MetricType:            "memoryused",
+		MetricType:            metricName,
 		StatWindowSeconds:     interval,
 		BreachDurationSeconds: interval,
 		Threshold:             threshold,
@@ -227,9 +227,9 @@ func generateDynamicScaleOutPolicy(instanceMin, instanceMax int, threshold int64
 	return string(bytes)
 }
 
-func generateDynamicScaleInPolicy(instanceMin, instanceMax int, threshold int64) string {
+func generateDynamicScaleInPolicy(instanceMin, instanceMax int, metricName string, threshold int64) string {
 	scalingInRule := ScalingRule{
-		MetricType:            "memoryused",
+		MetricType:            metricName,
 		StatWindowSeconds:     interval,
 		BreachDurationSeconds: interval,
 		Threshold:             threshold,

src/acceptance/app/dynamic_policy_test.go

@@ -38,7 +38,7 @@ var _ = Describe("AutoScaler dynamic policy", func() {
 		Expect(guid).To(Exit(0))
 		appGUID = strings.TrimSpace(string(guid.Out.Contents()))
 
-		Expect(cf.Cf("start", appName).Wait(cfg.DefaultTimeoutDuration() * 3)).To(Exit(0))
+		Expect(cf.Cf("start", appName).Wait(cfg.CfPushTimeoutDuration())).To(Exit(0))
 		waitForNInstancesRunning(appGUID, initialInstanceCount, cfg.DefaultTimeoutDuration())
 	})
 
@@ -49,7 +49,7 @@ var _ = Describe("AutoScaler dynamic policy", func() {
 		Expect(deleteService).To(Exit(0))
 	})
 
-	Context("when scale by memoryused", func() {
+	Context("when scaling by memoryused", func() {
 
 		JustBeforeEach(func() {
 			bindService := cf.Cf("bind-service", appName, instanceName, "-c", policy).Wait(cfg.DefaultTimeoutDuration())
@@ -63,12 +63,12 @@ var _ = Describe("AutoScaler dynamic policy", func() {
 
 		Context("when memory used is greater than scaling out threshold", func() {
 			BeforeEach(func() {
-				policy = generateDynamicScaleOutPolicy(1, 2, 30)
+				policy = generateDynamicScaleOutPolicy(1, 2, "memoryused", 30)
 				initialInstanceCount = 1
 			})
 
 			It("should scale out", func() {
-				totalTime := time.Duration(interval*2)*time.Second + 2*time.Minute
+				totalTime := time.Duration(interval*2)*time.Second + 3*time.Minute
 				finishTime := time.Now().Add(totalTime)
 
 				Eventually(func() uint64 {
@@ -82,11 +82,11 @@ var _ = Describe("AutoScaler dynamic policy", func() {
 
 		Context("when  memory used is lower than scaling in threshold", func() {
 			BeforeEach(func() {
-				policy = generateDynamicScaleInPolicy(1, 2, 80)
+				policy = generateDynamicScaleInPolicy(1, 2, "memoryused", 80)
 				initialInstanceCount = 2
 			})
 			It("should scale in", func() {
-				totalTime := time.Duration(interval*2)*time.Second + 2*time.Minute
+				totalTime := time.Duration(interval*2)*time.Second + 3*time.Minute
 				finishTime := time.Now().Add(totalTime)
 
 				Eventually(func() uint64 {
@@ -98,4 +98,54 @@ var _ = Describe("AutoScaler dynamic policy", func() {
 		})
 
 	})
+
+	Context("when scaling by memoryutil", func() {
+
+		JustBeforeEach(func() {
+			bindService := cf.Cf("bind-service", appName, instanceName, "-c", policy).Wait(cfg.DefaultTimeoutDuration())
+			Expect(bindService).To(Exit(0), "failed binding service to app with a policy ")
+		})
+
+		AfterEach(func() {
+			unbindService := cf.Cf("unbind-service", appName, instanceName).Wait(cfg.DefaultTimeoutDuration())
+			Expect(unbindService).To(Exit(0), "failed unbinding service from app")
+		})
+
+		Context("when memoryutil is greater than scaling out threshold", func() {
+			BeforeEach(func() {
+				policy = generateDynamicScaleOutPolicy(1, 2, "memoryutil", 20)
+				initialInstanceCount = 1
+			})
+
+			It("should scale out", func() {
+				totalTime := time.Duration(interval*2)*time.Second + 3*time.Minute
+				finishTime := time.Now().Add(totalTime)
+
+				Eventually(func() uint64 {
+					return averageMemoryUsedByInstance(appGUID, totalTime)
+				}, totalTime, 15*time.Second).Should(BeNumerically(">=", 26*MB))
+
+				waitForNInstancesRunning(appGUID, 2, finishTime.Sub(time.Now()))
+			})
+
+		})
+
+		Context("when  memoryutil is lower than scaling in threshold", func() {
+			BeforeEach(func() {
+				policy = generateDynamicScaleInPolicy(1, 2, "memoryutil", 80)
+				initialInstanceCount = 2
+			})
+			It("should scale in", func() {
+				totalTime := time.Duration(interval*2)*time.Second + 3*time.Minute
+				finishTime := time.Now().Add(totalTime)
+
+				Eventually(func() uint64 {
+					return averageMemoryUsedByInstance(appGUID, totalTime)
+				}, totalTime, 15*time.Second).Should(BeNumerically("<", 115*MB))
+
+				waitForNInstancesRunning(appGUID, 1, finishTime.Sub(time.Now()))
+			})
+		})
+
+	})
 })

src/acceptance/app/recurring_schedule_policy_test.go

@@ -48,11 +48,11 @@ var _ = Describe("AutoScaler recurring schedule policy", func() {
 		Expect(deleteService).To(Exit(0))
 	})
 
-	Context("when scale out by recurring schedule", func() {
+	Context("when scaling by recurring schedule", func() {
 
 		JustBeforeEach(func() {
 
-			Expect(cf.Cf("start", appName).Wait(cfg.DefaultTimeoutDuration() * 3)).To(Exit(0))
+			Expect(cf.Cf("start", appName).Wait(cfg.CfPushTimeoutDuration())).To(Exit(0))
 			waitForNInstancesRunning(appGUID, initialInstanceCount, cfg.DefaultTimeoutDuration())
 
 			location, err := time.LoadLocation("GMT")
@@ -75,12 +75,12 @@ var _ = Describe("AutoScaler recurring schedule policy", func() {
 			})
 
 			It("should scale", func() {
-				waitTime := startTime.Sub(time.Now()) + 1*time.Minute
 				By("setting to initial_min_instance_count")
-				waitForNInstancesRunning(appGUID, 3, waitTime)
+				jobRunTime := startTime.Add(1 * time.Minute).Sub(time.Now())
+				waitForNInstancesRunning(appGUID, 3, jobRunTime)
 
 				By("setting schedule's instance_min_count")
-				jobRunTime := endTime.Sub(time.Now())
+				jobRunTime = endTime.Sub(time.Now())
 				Eventually(func() int {
 					return runningInstances(appGUID, jobRunTime)
 				}, jobRunTime, 15*time.Second).Should(Equal(2))
@@ -102,13 +102,12 @@ var _ = Describe("AutoScaler recurring schedule policy", func() {
 			})
 
 			It("should scale", func() {
-				waitTime := startTime.Sub(time.Now()) + 1*time.Minute
-
 				By("setting to initial_min_instance_count")
-				waitForNInstancesRunning(appGUID, 3, waitTime)
+				jobRunTime := startTime.Add(1 * time.Minute).Sub(time.Now())
+				waitForNInstancesRunning(appGUID, 3, jobRunTime)
 
 				By("setting schedule's instance_min_count")
-				jobRunTime := endTime.Sub(time.Now())
+				jobRunTime = endTime.Sub(time.Now())
 				Eventually(func() int {
 					return runningInstances(appGUID, jobRunTime)
 				}, jobRunTime, 15*time.Second).Should(Equal(2))