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,49 @@ 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,
+		partitionid: p.ID,
+		sizeid:      machine.SizeID,
+	}
+
+	c := &s.PoolScalerConfig{
+		Log:       log.Named("pool-scaler"),
+		Manager:   manager,
+		Partition: *p,
+	}
+	scaler := s.NewPoolScaler(c)
+
+	config := states.StateConfig{
+		Log:       log.Named("fsm"),
+		Container: ec,
+		Event:     event,
+		Scaler:    scaler,
+		Machine:   machine,
+	}
+
+	newEC, err := fsm.HandleProvisioningEvent(&config)
 	if err != nil {
 		return nil, err
 	}
@@ -62,7 +92,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

@@ -21,6 +21,7 @@ type MachineSearchQuery struct {
 	Tags        []string `json:"tags" optional:"true"`
 
 	// allocation
+	NotAllocated        *bool       `json:"not_allocated" optional:"true"`
 	AllocationName      *string     `json:"allocation_name" optional:"true"`
 	AllocationProject   *string     `json:"allocation_project" optional:"true"`
 	AllocationImageID   *string     `json:"allocation_image_id" optional:"true"`
@@ -53,6 +54,7 @@ type MachineSearchQuery struct {
 	DiskSizes []int64  `json:"disk_sizes" optional:"true"`
 
 	// state
+	Sleeping   *bool   `json:"sleeping" optional:"true"`
 	StateValue *string `json:"state_value" optional:"true" enum:"|RESERVED|LOCKED"`
 
 	// ipmi
@@ -113,6 +115,10 @@ func (p *MachineSearchQuery) generateTerm(rs *RethinkStore) *r.Term {
 		})
 	}
 
+	if p.NotAllocated != nil && *p.NotAllocated {
+		q = q.Filter(map[string]any{"allocation": nil})
+	}
+
 	if p.AllocationName != nil {
 		q = q.Filter(func(row r.Term) r.Term {
 			return row.Field("allocation").Field("name").Eq(*p.AllocationName)

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)
+	notAllocated := true
+	q := MachineSearchQuery{
+		PartitionID:  &m.partitionid,
+		SizeID:       &m.sizeid,
+		StateValue:   &stateValue,
+		NotAllocated: &notAllocated,
+	}
+
+	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) {
+	sleeping := true
+	q := MachineSearchQuery{
+		PartitionID: &m.partitionid,
+		SizeID:      &m.sizeid,
+		Sleeping:    &sleeping,
+	}
+
+	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{
+		Sleeping:    true,
+		Description: "shut down as exceeding maximum partition poolsize",
+	}
+
+	err := m.rs.publishCommandAndUpdate(m.rs.log, machine, m.publisher, metal.MachineOffCmd, state)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (m *manager) PowerOn(machine *metal.Machine) error {
+	state := metal.MachineState{Sleeping: false}
+
+	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

@@ -150,13 +150,14 @@ func eventCallbacks(config *states.StateConfig) fsm.Callbacks {
 		// therefore we have an artificial state self_transition from which we can trigger the state-specific callback
 		"enter_" + SelfTransitionState: func(e *fsm.Event) {
 			if state, ok := allStates[e.Src]; ok {
-				state.OnTransition(e)
+				state.OnEnter(e)
 			}
 		},
 	}
 
 	for name, state := range allStates {
-		callbacks["enter_"+name] = state.OnTransition
+		callbacks["enter_"+name] = state.OnEnter
+		callbacks["leave_"+name] = state.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 {