auto-scaling available machine pool size

Makefile

@@ -43,3 +43,7 @@ visualize-fsm:
 	cd cmd/metal-api/internal/tools/visualize_fsm
 	go run main.go
 	dot -Tsvg fsm.dot > fsm.svg
+
+.PHONY: mocks
+mocks:
+	docker run --user $$(id -u):$$(id -g) --rm -w /work -v ${PWD}:/work vektra/mockery:v2.14.0 --name MachineManager --dir /work/cmd/metal-api/internal/scaler --output /work/cmd/metal-api/internal/scaler --filename pool_scaler_mock_test.go --testonly --inpackage

cmd/metal-api/internal/datastore/event.go

@@ -2,7 +2,10 @@ package datastore
 
 import (
 	"github.com/metal-stack/metal-api/cmd/metal-api/internal/fsm"
+	"github.com/metal-stack/metal-api/cmd/metal-api/internal/fsm/states"
 	"github.com/metal-stack/metal-api/cmd/metal-api/internal/metal"
+	s "github.com/metal-stack/metal-api/cmd/metal-api/internal/scaler"
+	"github.com/metal-stack/metal-lib/bus"
 	"go.uber.org/zap"
 )
 
@@ -37,22 +40,42 @@ func (rs *RethinkStore) CreateProvisioningEventContainer(ec *metal.ProvisioningE
 func (rs *RethinkStore) UpsertProvisioningEventContainer(ec *metal.ProvisioningEventContainer) error {
 	return rs.upsertEntity(rs.eventTable(), ec)
 }
-func (rs *RethinkStore) ProvisioningEventForMachine(log *zap.SugaredLogger, event *metal.ProvisioningEvent, machineID string) (*metal.ProvisioningEventContainer, error) {
-	ec, err := rs.FindProvisioningEventContainer(machineID)
+func (rs *RethinkStore) ProvisioningEventForMachine(log *zap.SugaredLogger, publisher bus.Publisher, event *metal.ProvisioningEvent, machine *metal.Machine) (*metal.ProvisioningEventContainer, error) {
+	ec, err := rs.FindProvisioningEventContainer(machine.ID)
 	if err != nil && !metal.IsNotFound(err) {
 		return nil, err
 	}
 
 	if ec == nil {
 		ec = &metal.ProvisioningEventContainer{
 			Base: metal.Base{
-				ID: machineID,
+				ID: machine.ID,
 			},
 			Liveliness: metal.MachineLivelinessAlive,
 		}
 	}
 
-	newEC, err := fsm.HandleProvisioningEvent(log, ec, event)
+	p, err := rs.FindPartition(machine.PartitionID)
+	if err != nil {
+		return nil, err
+	}
+
+	manager := &manager{
+		rs:        rs,
+		publisher: publisher,
+	}
+	scaler := s.NewPoolScaler(log, manager)
+
+	config := states.StateConfig{
+		Log:       log,
+		Container: ec,
+		Event:     event,
+		Scaler:    scaler,
+		Partition: p,
+		Machine:   machine,
+	}
+
+	newEC, err := fsm.HandleProvisioningEvent(&config)
 	if err != nil {
 		return nil, err
 	}
@@ -62,7 +85,10 @@ func (rs *RethinkStore) ProvisioningEventForMachine(log *zap.SugaredLogger, even
 	}
 
 	newEC.TrimEvents(100)
-
 	err = rs.UpsertProvisioningEventContainer(newEC)
+	if err != nil {
+		return nil, err
+	}
+
 	return newEC, err
 }

cmd/metal-api/internal/datastore/machine.go

@@ -58,6 +58,7 @@ type MachineSearchQuery struct {
 
 	// state
 	StateValue *string `json:"state_value" optional:"true"`
+	Waiting    *bool   `json:"waiting" optional:"true"`
 
 	// ipmi
 	IpmiAddress    *string `json:"ipmi_address" optional:"true"`

cmd/metal-api/internal/datastore/poolsize.go

@@ -0,0 +1,105 @@
+package datastore
+
+import (
+	e "github.com/metal-stack/metal-api/cmd/metal-api/internal/eventbus"
+	"github.com/metal-stack/metal-api/cmd/metal-api/internal/metal"
+	"github.com/metal-stack/metal-lib/bus"
+	"go.uber.org/zap"
+)
+
+type manager struct {
+	rs          *RethinkStore
+	publisher   bus.Publisher
+	partitionid string
+	sizeid      string
+}
+
+func (m *manager) AllMachines() (metal.Machines, error) {
+	q := MachineSearchQuery{
+		PartitionID: &m.partitionid,
+		SizeID:      &m.sizeid,
+	}
+
+	allMachines := metal.Machines{}
+	err := m.rs.SearchMachines(&q, &allMachines)
+	if err != nil {
+		return nil, err
+	}
+
+	return allMachines, nil
+}
+
+func (m *manager) WaitingMachines() (metal.Machines, error) {
+	stateValue := string(metal.AvailableState)
+	waiting := true
+	q := MachineSearchQuery{
+		PartitionID: &m.partitionid,
+		SizeID:      &m.sizeid,
+		StateValue:  &stateValue,
+		Waiting:     &waiting,
+	}
+
+	waitingMachines := metal.Machines{}
+	err := m.rs.SearchMachines(&q, &waitingMachines)
+	if err != nil {
+		return nil, err
+	}
+
+	return waitingMachines, nil
+}
+
+func (m *manager) ShutdownMachines() (metal.Machines, error) {
+	stateValue := string(metal.ShutdownState)
+	q := MachineSearchQuery{
+		PartitionID: &m.partitionid,
+		SizeID:      &m.sizeid,
+		StateValue:  &stateValue,
+	}
+
+	shutdownMachines := metal.Machines{}
+	err := m.rs.SearchMachines(&q, &shutdownMachines)
+	if err != nil {
+		return nil, err
+	}
+
+	return shutdownMachines, nil
+}
+
+func (m *manager) Shutdown(machine *metal.Machine) error {
+	state := metal.MachineState{
+		Value:       metal.ShutdownState,
+		Description: "shut down as exceeding maximum partition poolsize",
+	}
+
+	err := m.rs.publishCommandAndUpdate(m.rs.log, machine, m.publisher, metal.MachineOnCmd, state)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (m *manager) PowerOn(machine *metal.Machine) error {
+	state := metal.MachineState{Value: metal.AvailableState}
+
+	err := m.rs.publishCommandAndUpdate(m.rs.log, machine, m.publisher, metal.MachineOnCmd, state)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (rs *RethinkStore) publishCommandAndUpdate(logger *zap.SugaredLogger, m *metal.Machine, publisher bus.Publisher, cmd metal.MachineCommand, state metal.MachineState) error {
+	newMachine := *m
+	newMachine.State = state
+	err := rs.UpdateMachine(m, &newMachine)
+	if err != nil {
+		return err
+	}
+
+	err = e.PublishMachineCmd(logger, m, publisher, cmd)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

cmd/metal-api/internal/eventbus/nsq.go

@@ -109,3 +109,22 @@ func (n NSQClient) createTopics(partitions metal.Partitions, topics []metal.NSQT
 func (n NSQClient) delay() {
 	time.Sleep(nsqdRetryDelay)
 }
+
+func PublishMachineCmd(logger *zap.SugaredLogger, m *metal.Machine, publisher bus.Publisher, cmd metal.MachineCommand) error {
+	evt := metal.MachineEvent{
+		Type: metal.COMMAND,
+		Cmd: &metal.MachineExecCommand{
+			Command:         cmd,
+			TargetMachineID: m.ID,
+			IPMI:            &m.IPMI,
+		},
+	}
+
+	logger.Infow("publish event", "event", evt, "command", *evt.Cmd)
+	err := publisher.Publish(metal.TopicMachine.GetFQN(m.PartitionID), evt)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

cmd/metal-api/internal/fsm/events.go

@@ -159,5 +159,8 @@ func eventCallbacks(config *states.StateConfig) fsm.Callbacks {
 		callbacks["enter_"+name] = state.OnTransition
 	}
 
+	waiting := allStates[states.Waiting.String()].(*states.WaitingState)
+	callbacks["leave_"+states.Waiting.String()] = waiting.OnLeave
+
 	return callbacks
 }

cmd/metal-api/internal/fsm/fsm.go

@@ -8,65 +8,58 @@ import (
 	"github.com/looplab/fsm"
 	"github.com/metal-stack/metal-api/cmd/metal-api/internal/fsm/states"
 	"github.com/metal-stack/metal-api/cmd/metal-api/internal/metal"
-	"go.uber.org/zap"
 )
 
 // HandleProvisioningEvent can be called to determine whether the given incoming event follows an expected lifecycle of a machine considering the event history of the given provisioning event container.
 //
 // The function returns a new provisioning event container that can then be safely persisted in the database. If an error is returned, the incoming event is not supposed to be persisted in the database.
 //
 // Among other things, this function can detect crash loops or other irregularities within a machine lifecycle and enriches the returned provisioning event container with this information.
-func HandleProvisioningEvent(log *zap.SugaredLogger, ec *metal.ProvisioningEventContainer, event *metal.ProvisioningEvent) (*metal.ProvisioningEventContainer, error) {
-	if ec == nil {
-		return nil, fmt.Errorf("provisioning event container must not be nil")
-	}
-
-	if event == nil {
-		return nil, fmt.Errorf("provisioning event must not be nil")
+func HandleProvisioningEvent(c *states.StateConfig) (*metal.ProvisioningEventContainer, error) {
+	if err := c.Validate(); err != nil {
+		return nil, err
 	}
 
 	var (
-		clone     = *ec
-		container = &clone
-		f         = fsm.NewFSM(
-			initialStateFromEventContainer(container),
+		f = fsm.NewFSM(
+			initialStateFromEventContainer(c.Container),
 			Events(),
-			eventCallbacks(&states.StateConfig{Log: log, Event: event, Container: container}),
+			eventCallbacks(c),
 		)
 	)
 
-	err := f.Event(event.Event.String())
+	err := f.Event(c.Event.Event.String())
 	if err == nil {
-		return container, nil
+		return c.Container, nil
 	}
 
 	if errors.As(err, &fsm.InvalidEventError{}) {
-		if event.Message == "" {
-			event.Message = fmt.Sprintf("[unexpectedly received in %s]", strings.ToLower(f.Current()))
+		if c.Event.Message == "" {
+			c.Event.Message = fmt.Sprintf("[unexpectedly received in %s]", strings.ToLower(f.Current()))
 		} else {
-			event.Message = fmt.Sprintf("[unexpectedly received in %s]: %s", strings.ToLower(f.Current()), event.Message)
+			c.Event.Message = fmt.Sprintf("[unexpectedly received in %s]: %s", strings.ToLower(f.Current()), c.Event.Message)
 		}
 
-		container.LastEventTime = &event.Time
-		container.Liveliness = metal.MachineLivelinessAlive
-		container.LastErrorEvent = event
+		c.Container.LastEventTime = &c.Event.Time
+		c.Container.Liveliness = metal.MachineLivelinessAlive
+		c.Container.LastErrorEvent = c.Event
 
-		switch e := event.Event; e { //nolint:exhaustive
+		switch e := c.Event.Event; e { //nolint:exhaustive
 		case metal.ProvisioningEventPXEBooting, metal.ProvisioningEventPreparing:
-			container.CrashLoop = true
-			container.Events = append([]metal.ProvisioningEvent{*event}, container.Events...)
+			c.Container.CrashLoop = true
+			c.Container.Events = append([]metal.ProvisioningEvent{*c.Event}, c.Container.Events...)
 		case metal.ProvisioningEventAlive:
 			// under no circumstances we want to persists alive in the events container.
 			// when this happens the FSM gets stuck in invalid transitions
 			// (e.g. all following transitions are invalid and all subsequent alive events will be stored, cramping history).
 		default:
-			container.Events = append([]metal.ProvisioningEvent{*event}, container.Events...)
+			c.Container.Events = append([]metal.ProvisioningEvent{*c.Event}, c.Container.Events...)
 		}
 
-		return container, nil
+		return c.Container, nil
 	}
 
-	return nil, fmt.Errorf("internal error while calculating provisioning event container for machine %s: %w", container.ID, err)
+	return nil, fmt.Errorf("internal error while calculating provisioning event container for machine %s: %w", c.Container.ID, err)
 }
 
 func initialStateFromEventContainer(container *metal.ProvisioningEventContainer) string {

cmd/metal-api/internal/fsm/fsm_test.go

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/google/go-cmp/cmp"
+	"github.com/metal-stack/metal-api/cmd/metal-api/internal/fsm/states"
 	"github.com/metal-stack/metal-api/cmd/metal-api/internal/metal"
 	"go.uber.org/zap/zaptest"
 )
@@ -629,7 +630,13 @@ func TestHandleProvisioningEvent(t *testing.T) {
 	for i := range tests {
 		tt := tests[i]
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := HandleProvisioningEvent(zaptest.NewLogger(t).Sugar(), tt.container, tt.event)
+			params := states.StateConfig{
+				Log:       zaptest.NewLogger(t).Sugar(),
+				Container: tt.container,
+				Event:     tt.event,
+			}
+
+			got, err := HandleProvisioningEvent(&params)
 			if diff := cmp.Diff(tt.wantErr, err); diff != "" {
 				t.Errorf("HandleProvisioningEvent() diff = %s", diff)
 			}
@@ -649,15 +656,21 @@ func TestReactionToAllIncomingEvents(t *testing.T) {
 	// this test ensures that for every incoming event we have a proper transition
 	for e1 := range metal.AllProvisioningEventTypes {
 		for e2 := range metal.AllProvisioningEventTypes {
-			_, err := HandleProvisioningEvent(zaptest.NewLogger(t).Sugar(), &metal.ProvisioningEventContainer{
-				Events: metal.ProvisioningEvents{
-					{
-						Event: e2,
+			params := states.StateConfig{
+				Log: zaptest.NewLogger(t).Sugar(),
+				Container: &metal.ProvisioningEventContainer{
+					Events: metal.ProvisioningEvents{
+						{
+							Event: e2,
+						},
 					},
 				},
-			}, &metal.ProvisioningEvent{
-				Event: e1,
-			})
+				Event: &metal.ProvisioningEvent{
+					Event: e1,
+				},
+			}
+
+			_, err := HandleProvisioningEvent(&params)
 			if err != nil {
 				t.Errorf("transitioning from state %s with event %s: %s", e2, e1, err)
 			}

cmd/metal-api/internal/fsm/states/states.go

@@ -1,8 +1,11 @@
 package states
 
 import (
+	"fmt"
+
 	"github.com/looplab/fsm"
 	"github.com/metal-stack/metal-api/cmd/metal-api/internal/metal"
+	"github.com/metal-stack/metal-api/cmd/metal-api/internal/scaler"
 	"go.uber.org/zap"
 )
 
@@ -35,6 +38,21 @@ type StateConfig struct {
 	Log       *zap.SugaredLogger
 	Container *metal.ProvisioningEventContainer
 	Event     *metal.ProvisioningEvent
+	Scaler    *scaler.PoolScaler
+	Machine   *metal.Machine
+	Partition *metal.Partition
+}
+
+func (c *StateConfig) Validate() error {
+	if c.Container == nil {
+		return fmt.Errorf("provisioning event container must not be nil")
+	}
+
+	if c.Event == nil {
+		return fmt.Errorf("provisioning event must not be nil")
+	}
+
+	return nil
 }
 
 func AllStates(c *StateConfig) map[string]FSMState {