in pkg/monitor/sqsevent/asg-lifecycle-event.go [71:131]
func (m SQSMonitor) asgTerminationToInterruptionEvent(event *EventBridgeEvent, message *sqs.Message) (*monitor.InterruptionEvent, error) {
lifecycleDetail := &LifecycleDetail{}
err := json.Unmarshal(event.Detail, lifecycleDetail)
if err != nil {
return nil, err
}
if lifecycleDetail.Event == TEST_NOTIFICATION || lifecycleDetail.LifecycleTransition == TEST_NOTIFICATION {
return nil, skip{fmt.Errorf("message is an ASG test notification")}
}
nodeInfo, err := m.getNodeInfo(lifecycleDetail.EC2InstanceID)
if err != nil {
return nil, err
}
interruptionEvent := monitor.InterruptionEvent{
EventID: fmt.Sprintf("asg-lifecycle-term-%x", event.ID),
Kind: monitor.ASGLifecycleKind,
Monitor: SQSMonitorKind,
AutoScalingGroupName: lifecycleDetail.AutoScalingGroupName,
StartTime: event.getTime(),
NodeName: nodeInfo.Name,
IsManaged: nodeInfo.IsManaged,
InstanceID: lifecycleDetail.EC2InstanceID,
ProviderID: nodeInfo.ProviderID,
InstanceType: nodeInfo.InstanceType,
Description: fmt.Sprintf("ASG Lifecycle Termination event received. Instance will be interrupted at %s \n", event.getTime()),
}
stopHeartbeatCh := make(chan struct{})
interruptionEvent.PostDrainTask = func(interruptionEvent monitor.InterruptionEvent, _ node.Node) error {
_, err = m.continueLifecycleAction(lifecycleDetail)
if err != nil {
return fmt.Errorf("continuing ASG termination lifecycle: %w", err)
}
log.Info().Str("lifecycleHookName", lifecycleDetail.LifecycleHookName).Str("instanceID", lifecycleDetail.EC2InstanceID).Msg("Completed ASG Lifecycle Hook")
close(stopHeartbeatCh)
return m.deleteMessage(message)
}
interruptionEvent.PreDrainTask = func(interruptionEvent monitor.InterruptionEvent, n node.Node) error {
nthConfig := n.GetNthConfig()
// If only HeartbeatInterval is set, HeartbeatUntil will default to 172800.
if nthConfig.HeartbeatInterval != -1 && nthConfig.HeartbeatUntil != -1 {
go m.checkHeartbeatTimeout(nthConfig.HeartbeatInterval, lifecycleDetail)
go m.SendHeartbeats(nthConfig.HeartbeatInterval, nthConfig.HeartbeatUntil, lifecycleDetail, stopHeartbeatCh)
}
err := n.TaintASGLifecycleTermination(interruptionEvent.NodeName, interruptionEvent.EventID)
if err != nil {
log.Err(err).Msgf("unable to taint node with taint %s:%s", node.ASGLifecycleTerminationTaint, interruptionEvent.EventID)
}
return nil
}
return &interruptionEvent, nil
}