Add the ability to list only the latest scaling events.

cmd/scale/status/command.go

@@ -6,9 +6,9 @@ import (
 	"strings"
 
 	"github.com/jrasell/sherpa/cmd/helper"
-
 	"github.com/jrasell/sherpa/pkg/api"
 	clientCfg "github.com/jrasell/sherpa/pkg/config/client"
+	"github.com/jrasell/sherpa/pkg/config/scale"
 	"github.com/sean-/sysexits"
 	"github.com/spf13/cobra"
 )
@@ -27,6 +27,7 @@ func RegisterCommand(rootCmd *cobra.Command) error {
 		},
 	}
 	rootCmd.AddCommand(cmd)
+	scale.RegisterScaleStatusConfig(cmd)
 
 	return nil
 }
@@ -39,6 +40,7 @@ func runStatus(_ *cobra.Command, args []string) {
 
 	clientConfig := clientCfg.GetConfig()
 	mergedConfig := api.DefaultConfig(&clientConfig)
+	latestConfig := scale.GetScaleStatusConfig()
 
 	client, err := api.NewClient(mergedConfig)
 	if err != nil {
@@ -48,21 +50,20 @@ func runStatus(_ *cobra.Command, args []string) {
 
 	switch len(args) {
 	case 0:
-		os.Exit(runList(client))
+		os.Exit(runList(client, latestConfig.Latest))
 	case 1:
 		os.Exit(runInfo(client, args[0]))
 	}
 }
 
-func runList(c *api.Client) int {
-	resp, err := c.Scale().List()
+func runList(c *api.Client, latest bool) int {
+	resp, err := c.Scale().List(latest)
 	if err != nil {
 		fmt.Println("Error getting scaling list:", err)
 		os.Exit(sysexits.Software)
 	}
 
-	var out []string
-	out = append(out, listOutputHeader)
+	out := []string{listOutputHeader}
 
 	for id, jobEvents := range resp {
 		for jg, event := range jobEvents {

docs/api/scale.md

@@ -90,12 +90,17 @@ This endpoint can be used to list the recent scaling events.
 | :--------------------------- | :--------------------- |
 | `GET`    | `/v1/scale/status`              | `200 application/binary` |
 
+
+#### Parameters
+
+* `latest` (bool: optional) - Specifies whether Sherpa should only return the latest scaling event per job group.
+
 ### Sample Request
 
 ```
 $ curl \
     --request GET \
-    http://127.0.0.1:8000/v1/scale/status
+    http://127.0.0.1:8000/v1/scale/status?latest=false
 ```
 
 ### Sample Response
@@ -104,6 +109,7 @@ $ curl \
 {
   "036e4bd6-8f7d-4a8c-bf90-790790bbdc2a": {
     "example2:cache": {
+      "ID": "036e4bd6-8f7d-4a8c-bf90-790790bbdc2a",
       "EvalID": "e05a8d0f-87f8-bda8-eb3e-885caaf50c36",
       "Source": "InternalAutoscaler",
       "Time": 1568538833630403000,
@@ -119,6 +125,7 @@ $ curl \
   },
   "3bc8190e-b9fc-4997-bb39-3749eed5affd": {
     "example1:cache": {
+      "ID": "3bc8190e-b9fc-4997-bb39-3749eed5affd",
       "EvalID": "ec38990e-81e2-1c99-fbf2-725e8ca6ad70",
       "Source": "InternalAutoscaler",
       "Time": 1568538893629872000,
@@ -155,6 +162,7 @@ $ curl \
 ```json
 "3bc8190e-b9fc-4997-bb39-3749eed5affd": {
   "example1:cache": {
+    "ID": "3bc8190e-b9fc-4997-bb39-3749eed5affd",
     "EvalID": "ec38990e-81e2-1c99-fbf2-725e8ca6ad70",
     "Source": "InternalAutoscaler",
     "Time": 1568538893629872000,

pkg/api/api.go

@@ -108,12 +108,14 @@ func (c *Config) ConfigureTLS() error {
 	return nil
 }
 
-func (c *Client) get(endpoint string, out interface{}) error {
+func (c *Client) get(endpoint string, out interface{}, q *QueryOptions) error {
 	r, err := c.newRequest(http.MethodGet, endpoint)
 	if err != nil {
 		return err
 	}
 
+	r.setQueryOptions(q)
+
 	resp, err := c.doRequest(r)
 	resp, err = requireOK(resp, err, http.StatusOK)
 	if err != nil {

pkg/api/policy.go

@@ -28,7 +28,7 @@ type JobGroupPolicy struct {
 
 func (p *Policies) List() (*map[string]map[string]*JobGroupPolicy, error) {
 	var resp map[string]map[string]*JobGroupPolicy
-	err := p.client.get("/v1/policies", &resp)
+	err := p.client.get("/v1/policies", &resp, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -37,7 +37,7 @@ func (p *Policies) List() (*map[string]map[string]*JobGroupPolicy, error) {
 
 func (p *Policies) ReadJobPolicy(job string) (*map[string]*JobGroupPolicy, error) {
 	var resp map[string]*JobGroupPolicy
-	err := p.client.get("/v1/policy/"+job, &resp)
+	err := p.client.get("/v1/policy/"+job, &resp, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -49,7 +49,7 @@ func (p *Policies) ReadJobGroupPolicy(job, group string) (*JobGroupPolicy, error
 
 	path := fmt.Sprintf("/v1/policy/%s/%s", job, group)
 
-	err := p.client.get(path, &resp)
+	err := p.client.get(path, &resp, nil)
 	if err != nil {
 		return nil, err
 	}

pkg/api/scale.go

@@ -25,6 +25,7 @@ type ScaleResp struct {
 }
 
 type ScalingEvent struct {
+	ID      string
 	EvalID  string
 	Source  string
 	Time    int64
@@ -78,9 +79,12 @@ func (s *Scale) JobGroupIn(job, group string, count int, meta map[string]string)
 	return &resp, nil
 }
 
-func (s *Scale) List() (map[uuid.UUID]map[string]*ScalingEvent, error) {
+func (s *Scale) List(latest bool) (map[uuid.UUID]map[string]*ScalingEvent, error) {
 	var resp map[uuid.UUID]map[string]*ScalingEvent
-	err := s.client.get("/v1/scale/status", &resp)
+
+	q := QueryOptions{Params: map[string]string{"latest": strconv.FormatBool(latest)}}
+
+	err := s.client.get("/v1/scale/status", &resp, &q)
 	if err != nil {
 		return nil, err
 	}
@@ -89,7 +93,7 @@ func (s *Scale) List() (map[uuid.UUID]map[string]*ScalingEvent, error) {
 
 func (s *Scale) Info(id string) (map[string]*ScalingEvent, error) {
 	var resp map[string]*ScalingEvent
-	err := s.client.get("/v1/scale/status/"+id, &resp)
+	err := s.client.get("/v1/scale/status/"+id, &resp, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -100,6 +104,5 @@ func buildScaleReqBody(meta map[string]string) interface{} {
 	if meta == nil {
 		return nil
 	}
-
 	return &scaleReqBody{meta}
 }

pkg/api/system.go

@@ -32,7 +32,7 @@ func (c *Client) System() *System {
 
 func (s *System) Health() (*HealthResp, error) {
 	var resp HealthResp
-	err := s.client.get("/v1/system/health", &resp)
+	err := s.client.get("/v1/system/health", &resp, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -41,7 +41,7 @@ func (s *System) Health() (*HealthResp, error) {
 
 func (s *System) Info() (*InfoResp, error) {
 	var resp InfoResp
-	err := s.client.get("/v1/system/info", &resp)
+	err := s.client.get("/v1/system/info", &resp, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -50,7 +50,7 @@ func (s *System) Info() (*InfoResp, error) {
 
 func (s *System) Metrics() (*metrics.MetricsSummary, error) {
 	var resp metrics.MetricsSummary
-	err := s.client.get("/v1/system/metrics", &resp)
+	err := s.client.get("/v1/system/metrics", &resp, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -59,7 +59,7 @@ func (s *System) Metrics() (*metrics.MetricsSummary, error) {
 
 func (s *System) Leader() (*LeaderResp, error) {
 	var resp LeaderResp
-	err := s.client.get("/v1/system/leader", &resp)
+	err := s.client.get("/v1/system/leader", &resp, nil)
 	if err != nil {
 		return nil, err
 	}

pkg/config/scale/status.go

@@ -0,0 +1,37 @@
+package scale
+
+import (
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+)
+
+const (
+	configKeyScaleStatusLatest = "latest"
+)
+
+type StatusConfig struct {
+	Latest bool
+}
+
+func GetScaleStatusConfig() *StatusConfig {
+	return &StatusConfig{
+		Latest: viper.GetBool(configKeyScaleStatusLatest),
+	}
+}
+
+func RegisterScaleStatusConfig(cmd *cobra.Command) {
+	flags := cmd.PersistentFlags()
+
+	{
+		const (
+			key          = configKeyScaleStatusLatest
+			longOpt      = "latest"
+			defaultValue = false
+			description  = "List the latest scaling event for each job group only"
+		)
+
+		flags.Bool(longOpt, defaultValue, description)
+		_ = viper.BindPFlag(key, flags.Lookup(longOpt))
+		viper.SetDefault(key, defaultValue)
+	}
+}

pkg/config/scale/status_test.go

@@ -0,0 +1,16 @@
+package scale
+
+import (
+	"testing"
+
+	"github.com/spf13/cobra"
+	"github.com/stretchr/testify/assert"
+)
+
+func Test_ScaleStatusConfig(t *testing.T) {
+	fakeCMD := &cobra.Command{}
+	RegisterScaleConfig(fakeCMD)
+
+	cfg := GetScaleStatusConfig()
+	assert.Equal(t, false, cfg.Latest)
+}

pkg/scale/v1/status.go

@@ -6,9 +6,15 @@ import (
 
 	"github.com/gofrs/uuid"
 	"github.com/gorilla/mux"
+	"github.com/jrasell/sherpa/pkg/state"
 )
 
 func (s *Scale) StatusList(w http.ResponseWriter, r *http.Request) {
+	if l := r.URL.Query().Get("latest"); l == "true" {
+		s.statusListLatest(w)
+		return
+	}
+
 	list, err := s.stateBackend.GetScalingEvents()
 	if err != nil {
 		s.logger.Error().Err(err).Msg("failed to get scaling events from state")
@@ -26,6 +32,29 @@ func (s *Scale) StatusList(w http.ResponseWriter, r *http.Request) {
 	writeJSONResponse(w, bytes, http.StatusOK)
 }
 
+func (s *Scale) statusListLatest(w http.ResponseWriter) {
+	list, err := s.stateBackend.GetLatestScalingEvents()
+	if err != nil {
+		s.logger.Error().Err(err).Msg("failed to get latest scaling events from state")
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	out := map[uuid.UUID]map[string]*state.ScalingEvent{}
+	for jg, event := range list {
+		out[event.ID] = map[string]*state.ScalingEvent{jg: event}
+	}
+
+	bytes, err := json.Marshal(out)
+	if err != nil {
+		s.logger.Error().Err(err).Msg("failed to marshal latest scaling state response")
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	writeJSONResponse(w, bytes, http.StatusOK)
+}
+
 func (s *Scale) StatusInfo(w http.ResponseWriter, r *http.Request) {
 	vars := mux.Vars(r)
 	id := vars["id"]

pkg/server/config.go

@@ -17,11 +17,10 @@ const (
 	routeUIName            = "UI"
 	routeUIPattern         = "/ui"
 
-	routeGetScalingStatusPattern = "/v1/scale/status"
-	routeGetScalingStatusName    = "GetScalingStatus"
-	routeGetScalingInfoPattern   = "/v1/scale/status/{id}"
-	routeGetScalingInfoName      = "GetScalingInfo"
-
+	routeGetScalingStatusPattern            = "/v1/scale/status"
+	routeGetScalingStatusName               = "GetScalingStatus"
+	routeGetScalingInfoPattern              = "/v1/scale/status/{id}"
+	routeGetScalingInfoName                 = "GetScalingInfo"
 	routeScaleOutJobGroupName               = "ScaleOutJobGroup"
 	routeScaleOutJobGroupPattern            = "/v1/scale/out/{job_id}/{group}"
 	routeScaleInJobGroupName                = "ScaleInJobGroup"

pkg/server/routes.go

@@ -114,7 +114,6 @@ func (h *HTTPServer) setupScaleRoutes() []router.Route {
 			Pattern: routeGetScalingStatusPattern,
 			Handler: leaderProtectedHandler(h.clusterMember, h.routes.Scale.StatusList),
 		},
-
 		router.Route{
 			Name:    routeGetScalingInfoName,
 			Method:  http.MethodGet,

pkg/state/scale.go

@@ -21,6 +21,9 @@ type ScalingState struct {
 // ScalingEvent represents a single scaling event state entry that is persisted to the backend
 // store.
 type ScalingEvent struct {
+	// ID is the scaling ID.
+	ID uuid.UUID
+
 	// EvalID is the Nomad evaluation ID which was created as a result of submitting the updated
 	// job to the Nomad API.
 	EvalID string

pkg/state/scale/consul/consul.go

@@ -169,6 +169,7 @@ func (s StateBackend) PutScalingEvent(job string, event *state.ScalingEventMessa
 	defer metrics.MeasureSince(metricKeyPutEvent, time.Now())
 
 	sEntry := &state.ScalingEvent{
+		ID:      event.ID,
 		EvalID:  event.EvalID,
 		Source:  event.Source,
 		Time:    event.Time,

pkg/state/scale/memory/memory.go

@@ -74,6 +74,7 @@ func (s *StateBackend) PutScalingEvent(job string, event *state.ScalingEventMess
 	k := job + ":" + event.GroupName
 
 	sEntry := &state.ScalingEvent{
+		ID:      event.ID,
 		EvalID:  event.EvalID,
 		Source:  event.Source,
 		Time:    event.Time,

pkg/state/scale/memory/memory_test.go

@@ -121,6 +121,7 @@ func generateTestEvent(t int64) *state.ScalingEventMessage {
 
 func convertMessageToStateRepresentation(event *state.ScalingEventMessage) *state.ScalingEvent {
 	return &state.ScalingEvent{
+		ID:      event.ID,
 		EvalID:  event.EvalID,
 		Source:  event.Source,
 		Time:    event.Time,