diff --git a/pkg/datapath/micro_service_watcher.go b/pkg/datapath/micro_service_watcher.go index 7f68a570f..a824ea469 100644 --- a/pkg/datapath/micro_service_watcher.go +++ b/pkg/datapath/micro_service_watcher.go @@ -19,6 +19,7 @@ package datapath import ( "context" "encoding/json" + "fmt" "os" "strings" "sync" @@ -55,6 +56,7 @@ const ( EventReasonProgress = "Data-Path-Progress" EventReasonCancelling = "Data-Path-Canceling" EventReasonStopped = "Data-Path-Stopped" + EventReasonEvicted = "Evicted" ) type microServiceBRWatcher struct { @@ -81,6 +83,7 @@ type microServiceBRWatcher struct { eventHandler cache.ResourceEventHandlerRegistration podHandler cache.ResourceEventHandlerRegistration watcherLock sync.Mutex + eventMessages sync.Map } func newMicroServiceBRWatcher(client client.Client, kubeClient kubernetes.Interface, mgr manager.Manager, taskType string, taskName string, namespace string, @@ -329,6 +332,8 @@ func (ms *microServiceBRWatcher) startWatch() { ms.callbacks.OnCancelled(ms.ctx, ms.namespace, ms.taskName) } else if terminateMessage != "" { ms.callbacks.OnFailed(ms.ctx, ms.namespace, ms.taskName, errors.New(terminateMessage)) + } else if msg, evicted := ms.eventMessages.Load(EventReasonEvicted); evicted { + ms.callbacks.OnFailed(ms.ctx, ms.namespace, ms.taskName, errors.New(msg.(string))) } else { ms.callbacks.OnFailed(ms.ctx, ms.namespace, ms.taskName, errors.New(lastPod.Status.Message)) } @@ -356,6 +361,9 @@ func (ms *microServiceBRWatcher) onEvent(evt *corev1api.Event) { case EventReasonStopped: ms.terminatedFromEvent.Store(true) ms.log.Infof("Received data path stop message: %s", evt.Message) + case EventReasonEvicted: + ms.eventMessages.Store(EventReasonEvicted, fmt.Sprintf("data path pod was evicted, message: %s", evt.Message)) + ms.log.Infof("Pod was evicted for data path %s, message: %s", ms.taskName, evt.Message) default: ms.log.Infof("Received event for data path %s, reason: %s, message: %s", ms.taskName, evt.Reason, evt.Message) }