Merge pull request #282 from hashicorp/f-eval-broker

Evaluation broker

.circleci/config/jobs/lint-go.yml

@@ -3,4 +3,5 @@ steps:
   - install-go
   - install-golangci-lint
   - checkout
+  - run: make check
   - run: make lint

agent/agent.go

@@ -6,16 +6,18 @@ import (
 	"os"
 	"os/signal"
 	"syscall"
+	"time"
 
 	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/nomad-autoscaler/agent/config"
 	agentServer "github.com/hashicorp/nomad-autoscaler/agent/http"
-	nomadHelper "github.com/hashicorp/nomad-autoscaler/helper/nomad"
 	"github.com/hashicorp/nomad-autoscaler/plugins/manager"
 	"github.com/hashicorp/nomad-autoscaler/policy"
 	filePolicy "github.com/hashicorp/nomad-autoscaler/policy/file"
 	nomadPolicy "github.com/hashicorp/nomad-autoscaler/policy/nomad"
+	"github.com/hashicorp/nomad-autoscaler/policyeval"
 	"github.com/hashicorp/nomad-autoscaler/sdk"
+	nomadHelper "github.com/hashicorp/nomad-autoscaler/sdk/helper/nomad"
 	"github.com/hashicorp/nomad/api"
 )
 
@@ -26,6 +28,7 @@ type Agent struct {
 	pluginManager *manager.PluginManager
 	policyManager *policy.Manager
 	httpServer    *agentServer.Server
+	evalBroker    *policyeval.Broker
 }
 
 func NewAgent(c *config.Agent, logger hclog.Logger) *Agent {
@@ -70,6 +73,10 @@ func (a *Agent) Run() error {
 	policyEvalCh := a.setupPolicyManager()
 	go a.policyManager.Run(ctx, policyEvalCh)
 
+	// Launch eval broker and workers.
+	a.evalBroker = policyeval.NewBroker(a.logger.ResetNamed("policy_eval"), 5*time.Minute, 3)
+	a.initWorkers(ctx)
+
 	// Launch the eval handler.
 	go a.runEvalHandler(ctx, policyEvalCh)
 
@@ -85,12 +92,27 @@ func (a *Agent) runEvalHandler(ctx context.Context, evalCh chan *sdk.ScalingEval
 			a.logger.Info("context closed, shutting down eval handler")
 			return
 		case policyEval := <-evalCh:
-			w := policy.NewWorker(a.logger, a.pluginManager, a.policyManager)
-			go w.HandlePolicy(ctx, policyEval)
+			a.evalBroker.Enqueue(policyEval)
 		}
 	}
 }
 
+func (a *Agent) initWorkers(ctx context.Context) {
+	policyEvalLogger := a.logger.ResetNamed("policy_eval")
+
+	for i := 0; i < a.config.PolicyWorkers.Horizontal; i++ {
+		w := policyeval.NewBaseWorker(
+			policyEvalLogger, a.pluginManager, a.policyManager, a.evalBroker, "horizontal")
+		go w.Run(ctx)
+	}
+
+	for i := 0; i < a.config.PolicyWorkers.Cluster; i++ {
+		w := policyeval.NewBaseWorker(
+			policyEvalLogger, a.pluginManager, a.policyManager, a.evalBroker, "cluster")
+		go w.Run(ctx)
+	}
+}
+
 func (a *Agent) setupPolicyManager() chan *sdk.ScalingEvaluation {
 
 	// Create our processor, a shared method for performing basic policy

agent/config/config.go

@@ -8,8 +8,8 @@ import (
 	"time"
 
 	"github.com/hashicorp/hcl/v2/hclsimple"
-	"github.com/hashicorp/nomad-autoscaler/helper/file"
 	"github.com/hashicorp/nomad-autoscaler/plugins"
+	"github.com/hashicorp/nomad-autoscaler/sdk/helper/file"
 	"github.com/mitchellh/copystructure"
 )
 
@@ -43,6 +43,10 @@ type Agent struct {
 	// Policy is the configuration used to setup the policy manager.
 	Policy *Policy `hcl:"policy,block"`
 
+	// PolicyWorkers is the configuration used to define the number of workers
+	// to start for each policy type.
+	PolicyWorkers *PolicyWorkers `hcl:"policy_workers,block"`
+
 	// Telemetry is the configuration used to setup metrics collection.
 	Telemetry *Telemetry `hcl:"telemetry,block"`
 
@@ -247,6 +251,16 @@ type Policy struct {
 	DefaultEvaluationIntervalHCL string `hcl:"default_evaluation_interval,optional" json:"-"`
 }
 
+// PolicyWorkers holds the configuration for the number of policy evaluation
+// workers for each policy type.
+type PolicyWorkers struct {
+	ClusterPtr *int `hcl:"cluster,optional" json:"-"`
+	Cluster    int
+
+	HorizontalPtr *int `hcl:"horizontal,optional" json:"-"`
+	Horizontal    int
+}
+
 const (
 	// defaultLogLevel is the default log level used for the Autoscaler agent.
 	defaultLogLevel = "info"
@@ -280,6 +294,14 @@ const (
 	// defaultTelemetryCollectionInterval is the default telemetry metrics
 	// collection interval.
 	defaultTelemetryCollectionInterval = 1 * time.Second
+
+	// defaultClusterPolicyWorkers is the default number of workers for cluster
+	// policies.
+	defaultClusterPolicyWorkers = 10
+
+	// defaultHorizontalPolicyWorkers is the default number of workers for
+	// horizontal policies.
+	defaultHorizontalPolicyWorkers = 10
 )
 
 // Default is used to generate a new default agent configuration.
@@ -310,6 +332,10 @@ func Default() (*Agent, error) {
 			DefaultCooldown:           defaultPolicyCooldown,
 			DefaultEvaluationInterval: defaultEvaluationInterval,
 		},
+		PolicyWorkers: &PolicyWorkers{
+			Cluster:    defaultClusterPolicyWorkers,
+			Horizontal: defaultHorizontalPolicyWorkers,
+		},
 		APMs:       []*Plugin{{Name: plugins.InternalAPMNomad, Driver: plugins.InternalAPMNomad}},
 		Strategies: []*Plugin{{Name: plugins.InternalStrategyTargetValue, Driver: plugins.InternalStrategyTargetValue}},
 		Targets:    []*Plugin{{Name: plugins.InternalTargetNomad, Driver: plugins.InternalTargetNomad}},
@@ -345,6 +371,10 @@ func (a *Agent) Merge(b *Agent) *Agent {
 		result.Policy = result.Policy.merge(b.Policy)
 	}
 
+	if b.PolicyWorkers != nil {
+		result.PolicyWorkers = result.PolicyWorkers.merge(b.PolicyWorkers)
+	}
+
 	if len(result.APMs) == 0 && len(b.APMs) != 0 {
 		apmCopy := make([]*Plugin, len(b.APMs))
 		for i, v := range b.APMs {
@@ -542,6 +572,22 @@ func (p *Policy) merge(b *Policy) *Policy {
 	return &result
 }
 
+func (pw *PolicyWorkers) merge(in *PolicyWorkers) *PolicyWorkers {
+	result := *pw
+
+	if in.ClusterPtr != nil {
+		result.ClusterPtr = in.ClusterPtr
+		result.Cluster = in.Cluster
+	}
+
+	if in.HorizontalPtr != nil {
+		result.HorizontalPtr = in.HorizontalPtr
+		result.Horizontal = in.Horizontal
+	}
+
+	return &result
+}
+
 // pluginConfigSetMerge merges two sets of plugin configs. For plugins with the
 // same name, the configs are merged.
 func pluginConfigSetMerge(first, second []*Plugin) []*Plugin {
@@ -621,6 +667,15 @@ func parseFile(file string, cfg *Agent) error {
 		}
 	}
 
+	if cfg.PolicyWorkers != nil {
+		if cfg.PolicyWorkers.ClusterPtr != nil {
+			cfg.PolicyWorkers.Cluster = *cfg.PolicyWorkers.ClusterPtr
+		}
+		if cfg.PolicyWorkers.HorizontalPtr != nil {
+			cfg.PolicyWorkers.Horizontal = *cfg.PolicyWorkers.HorizontalPtr
+		}
+	}
+
 	return nil
 }
 

agent/config/config_test.go

@@ -8,6 +8,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/hashicorp/nomad-autoscaler/sdk/helper/ptr"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -23,6 +24,8 @@ func Test_Default(t *testing.T) {
 	assert.Equal(t, "127.0.0.1", def.HTTP.BindAddress)
 	assert.Equal(t, 8080, def.HTTP.BindPort)
 	assert.Equal(t, def.Policy.DefaultCooldown, 5*time.Minute)
+	assert.Equal(t, defaultClusterPolicyWorkers, def.PolicyWorkers.Cluster)
+	assert.Equal(t, defaultHorizontalPolicyWorkers, def.PolicyWorkers.Horizontal)
 	assert.Len(t, def.APMs, 1)
 	assert.Len(t, def.Targets, 1)
 	assert.Len(t, def.Strategies, 1)
@@ -41,6 +44,10 @@ func TestAgent_Merge(t *testing.T) {
 		Nomad: &Nomad{
 			Address: "http://nomad.systems:4646",
 		},
+		PolicyWorkers: &PolicyWorkers{
+			HorizontalPtr: ptr.IntToPtr(5),
+			Horizontal:    5,
+		},
 		APMs: []*Plugin{
 			{
 				Name:   "prometheus",
@@ -75,6 +82,12 @@ func TestAgent_Merge(t *testing.T) {
 			DefaultCooldown:           20 * time.Minute,
 			DefaultEvaluationInterval: 10 * time.Second,
 		},
+		PolicyWorkers: &PolicyWorkers{
+			ClusterPtr:    ptr.IntToPtr(8),
+			Cluster:       8,
+			HorizontalPtr: ptr.IntToPtr(7),
+			Horizontal:    7,
+		},
 		Telemetry: &Telemetry{
 			StatsiteAddr:                       "some-address",
 			StatsdAddr:                         "some-other-address",
@@ -144,6 +157,12 @@ func TestAgent_Merge(t *testing.T) {
 			DefaultCooldown:           20 * time.Minute,
 			DefaultEvaluationInterval: 10 * time.Second,
 		},
+		PolicyWorkers: &PolicyWorkers{
+			Cluster:       8,
+			ClusterPtr:    ptr.IntToPtr(8),
+			Horizontal:    7,
+			HorizontalPtr: ptr.IntToPtr(7),
+		},
 		Telemetry: &Telemetry{
 			StatsiteAddr:                       "some-address",
 			StatsdAddr:                         "some-other-address",
@@ -211,6 +230,7 @@ func TestAgent_Merge(t *testing.T) {
 	assert.Equal(t, expectedResult.Nomad, actualResult.Nomad)
 	assert.Equal(t, expectedResult.PluginDir, actualResult.PluginDir)
 	assert.Equal(t, expectedResult.Policy, actualResult.Policy)
+	assert.Equal(t, expectedResult.PolicyWorkers, actualResult.PolicyWorkers)
 	assert.ElementsMatch(t, expectedResult.APMs, actualResult.APMs)
 	assert.ElementsMatch(t, expectedResult.Targets, actualResult.Targets)
 	assert.ElementsMatch(t, expectedResult.Strategies, actualResult.Strategies)

agent/plugins.go

@@ -4,9 +4,9 @@ import (
 	"strconv"
 
 	"github.com/hashicorp/nomad-autoscaler/agent/config"
-	nomadHelper "github.com/hashicorp/nomad-autoscaler/helper/nomad"
 	"github.com/hashicorp/nomad-autoscaler/plugins"
 	"github.com/hashicorp/nomad-autoscaler/plugins/manager"
+	nomadHelper "github.com/hashicorp/nomad-autoscaler/sdk/helper/nomad"
 )
 
 // setupPlugins is used to setup the plugin manager for all the agents plugins

command/agent.go

@@ -9,7 +9,7 @@ import (
 	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/nomad-autoscaler/agent"
 	"github.com/hashicorp/nomad-autoscaler/agent/config"
-	flaghelper "github.com/hashicorp/nomad-autoscaler/helper/flag"
+	flaghelper "github.com/hashicorp/nomad-autoscaler/sdk/helper/flag"
 )
 
 type AgentCommand struct {
@@ -188,7 +188,7 @@ Telemetry Options:
 
   -telemetry-circonus-broker-id
     The Circonus broker to use when creating a new check.
-  
+
   -telemetry-circonus-broker-select-tag
     A tag which is used to select a broker ID when an explicit broker ID is not
     provided.

go.mod

@@ -16,7 +16,7 @@ require (
 	github.com/hashicorp/go-multierror v1.0.0
 	github.com/hashicorp/go-plugin v1.0.1
 	github.com/hashicorp/hcl/v2 v2.3.0
-	github.com/hashicorp/nomad/api v0.0.0-20200909222706-48e6706e689b
+	github.com/hashicorp/nomad/api v0.0.0-20200929215746-5a48a8d725b0
 	github.com/kr/pretty v0.2.0 // indirect
 	github.com/mattn/go-isatty v0.0.12 // indirect
 	github.com/mitchellh/cli v1.0.0

go.sum

@@ -118,8 +118,8 @@ github.com/hashicorp/golang-lru v0.5.0 h1:CL2msUPvZTLb5O648aiLNJw3hnBxN2+1Jq8rCO
 github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/hcl/v2 v2.3.0 h1:iRly8YaMwTBAKhn1Ybk7VSdzbnopghktCD031P8ggUE=
 github.com/hashicorp/hcl/v2 v2.3.0/go.mod h1:d+FwDBbOLvpAM3Z6J7gPj/VoAGkNe/gm352ZhjJ/Zv8=
-github.com/hashicorp/nomad/api v0.0.0-20200909222706-48e6706e689b h1:mUI70KkI+e3rD7dJAEVhC12mIYU04WC1ICviM4DIbhg=
-github.com/hashicorp/nomad/api v0.0.0-20200909222706-48e6706e689b/go.mod h1:DCi2k47yuUDzf2qWAK8E1RVmWgz/lc0jZQeEnICTxmY=
+github.com/hashicorp/nomad/api v0.0.0-20200929215746-5a48a8d725b0 h1:kYP+3yZA2/QEDDNpsGPb9c+X52P6EA4FBsKKE29ZsBU=
+github.com/hashicorp/nomad/api v0.0.0-20200929215746-5a48a8d725b0/go.mod h1:DCi2k47yuUDzf2qWAK8E1RVmWgz/lc0jZQeEnICTxmY=
 github.com/hashicorp/yamux v0.0.0-20180604194846-3520598351bb h1:b5rjCoWHc7eqmAS4/qyk21ZsHyb6Mxv/jykxvNTkU4M=
 github.com/hashicorp/yamux v0.0.0-20180604194846-3520598351bb/go.mod h1:+NfK9FKeTrX5uv1uIXGdwYDTeHna2qgaIlx54MXqjAM=
 github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af h1:pmfjZENx5imkbgOkpRUYLnmbU7UEFbjtDA2hxJ1ichM=

plugins/builtin/apm/nomad/plugin/node.go

@@ -6,8 +6,8 @@ import (
 	"strings"
 	"time"
 
-	"github.com/hashicorp/nomad-autoscaler/helper/scaleutils"
 	"github.com/hashicorp/nomad-autoscaler/sdk"
+	"github.com/hashicorp/nomad-autoscaler/sdk/helper/scaleutils"
 	"github.com/hashicorp/nomad/api"
 )
 

plugins/builtin/apm/nomad/plugin/node_test.go

@@ -4,7 +4,7 @@ import (
 	"errors"
 	"testing"
 
-	"github.com/hashicorp/nomad-autoscaler/helper/scaleutils"
+	"github.com/hashicorp/nomad-autoscaler/sdk/helper/scaleutils"
 	"github.com/hashicorp/nomad/api"
 	"github.com/stretchr/testify/assert"
 )

plugins/builtin/apm/nomad/plugin/plugin.go

@@ -4,10 +4,10 @@ import (
 	"fmt"
 
 	hclog "github.com/hashicorp/go-hclog"
-	nomadHelper "github.com/hashicorp/nomad-autoscaler/helper/nomad"
 	"github.com/hashicorp/nomad-autoscaler/plugins"
 	"github.com/hashicorp/nomad-autoscaler/plugins/apm"
 	"github.com/hashicorp/nomad-autoscaler/plugins/base"
+	nomadHelper "github.com/hashicorp/nomad-autoscaler/sdk/helper/nomad"
 	"github.com/hashicorp/nomad/api"
 )
 

plugins/builtin/target/aws-asg/plugin/aws.go

@@ -9,8 +9,8 @@ import (
 	"github.com/aws/aws-sdk-go-v2/aws/external"
 	"github.com/aws/aws-sdk-go-v2/service/autoscaling"
 	"github.com/aws/aws-sdk-go-v2/service/ec2"
-	"github.com/hashicorp/nomad-autoscaler/helper/scaleutils"
 	"github.com/hashicorp/nomad-autoscaler/sdk"
+	"github.com/hashicorp/nomad-autoscaler/sdk/helper/scaleutils"
 )
 
 const (

plugins/builtin/target/aws-asg/plugin/aws_test.go

@@ -5,7 +5,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/hashicorp/nomad-autoscaler/helper/scaleutils"
+	"github.com/hashicorp/nomad-autoscaler/sdk/helper/scaleutils"
 	"github.com/stretchr/testify/assert"
 )
 

plugins/builtin/target/aws-asg/plugin/plugin.go

@@ -8,12 +8,12 @@ import (
 	"github.com/aws/aws-sdk-go-v2/service/autoscaling"
 	"github.com/aws/aws-sdk-go-v2/service/ec2"
 	hclog "github.com/hashicorp/go-hclog"
-	"github.com/hashicorp/nomad-autoscaler/helper/nomad"
-	"github.com/hashicorp/nomad-autoscaler/helper/scaleutils"
 	"github.com/hashicorp/nomad-autoscaler/plugins"
 	"github.com/hashicorp/nomad-autoscaler/plugins/base"
 	"github.com/hashicorp/nomad-autoscaler/plugins/target"
 	"github.com/hashicorp/nomad-autoscaler/sdk"
+	"github.com/hashicorp/nomad-autoscaler/sdk/helper/nomad"
+	"github.com/hashicorp/nomad-autoscaler/sdk/helper/scaleutils"
 )
 
 const (

plugins/builtin/target/nomad/plugin/plugin.go

@@ -6,11 +6,11 @@ import (
 	"time"
 
 	hclog "github.com/hashicorp/go-hclog"
-	nomadHelper "github.com/hashicorp/nomad-autoscaler/helper/nomad"
 	"github.com/hashicorp/nomad-autoscaler/plugins"
 	"github.com/hashicorp/nomad-autoscaler/plugins/base"
 	"github.com/hashicorp/nomad-autoscaler/plugins/target"
 	"github.com/hashicorp/nomad-autoscaler/sdk"
+	nomadHelper "github.com/hashicorp/nomad-autoscaler/sdk/helper/nomad"
 	"github.com/hashicorp/nomad/api"
 )