Updating resources upon relative to pod update

docs/configuration.md

@@ -20,7 +20,7 @@ for automation.
 Even though most use-cases are covered by either planned or direct emulation, some edge-cases might require both.
 However, this is not supported in the initial version of Apate, and might lead to unexpected results in some cases, 
 due to CRD resync.  
-We invite others to contribute to Apate and add this feature, as it should be a good first issue.
+We invite others to contribute to Apate and add this feature, as it should be a [good first issue](https://github.com/atlarge-research/apate/issues/5).
 :::  
 
 ## Nodes
@@ -90,7 +90,7 @@ State is the desired state of the node.
 ::: warning  
 In the initial version of Apate, it is not possible to revert `node_failed` or `heartbeat_failed` directly. 
 However, this can still be achieved using a custom state, and reverting the timout responses manually.   
-We invite others to contribute to Apate and add this feature, as it should be a good first issue.  
+We invite others to contribute to Apate and add this feature, as it should be a [good first issue](https://github.com/atlarge-research/apate/issues/6).
 :::
 
 #### Custom state
@@ -152,7 +152,7 @@ Task is a combination of a timestamp and a state
 | Field | Type | Description | Required |
 | --- | --- | --- | --- |
 | timestamp | [Time](#time) | Time at which this task will be executed | Yes |
-| relative_to_pod | bool | If true, the timestamp will be relative to the start time of the pod, instead of the start time of the scenario | No |
+| relative_to_pod | bool | If true, the timestamp will be relative to the start time of the pod, instead of the start time of the scenario. Note that currently, if this setting is used once for a certain flag, further updates of this flag relative to the scenario will not be used. This is a limitation we aim to solve [in the future](https://github.com/atlarge-research/apate/issues/7). | No |
 | state | [State](#pod-state) | Desired state after this task | Yes |
 
 ### Pod state

services/apatelet/store/get_flag.go

@@ -39,11 +39,11 @@ func (s *store) GetPodFlag(pod *corev1.Pod, flag events.PodEventFlag) (interface
 
 	label, ok := getPodLabelByPod(pod)
 	if ok {
-		if val, ok := s.podFlags[label][flag]; ok {
+		if val, ok := s.getPodTimeFlag(pod, flag, label); ok {
 			return val, nil
 		}
 
-		if val, ok := s.getPodTimeFlag(pod, flag, label); ok {
+		if val, ok := s.podFlags[label][flag]; ok {
 			return val, nil
 		}
 	}
@@ -92,20 +92,33 @@ func (s *store) getPodTimeFlag(pod *corev1.Pod, flag events.PodEventFlag, label
 			// Look at the previous index
 			currentPodFlags := timeFlags[previousIndex]
 
-			// If this index has time flags before now (it might not have if this is the first iteration)
-			if podStartTime.Add(currentPodFlags.TimeSincePodStart).Before(time.Now()) {
-				if pf, ok := currentPodFlags.Flags[flag]; ok {
-					// Set cache and return it
-					s.podTimeIndexCache[pod][flag] = previousIndex
+			if pf, ok := currentPodFlags.Flags[flag]; ok {
+				if podStartTime.Add(currentPodFlags.TimeSincePodStart).Before(time.Now()) {
+					// If this index has time flags before now (it might not have if this is the first iteration)
+					s.updateCache(pf, pod, flag, previousIndex)
 					return pf, true
 				}
 			}
 
-			// Else set the current index, as the next iteration can skip every index thus far
-			s.podTimeIndexCache[pod][flag] = i
 			break
 		}
 	}
 
 	return nil, false
 }
+
+func (s *store) updateCache(flagValue interface{}, pod *corev1.Pod, flag events.EventFlag, newIndex int) {
+	prevIndex, ok := s.podTimeIndexCache[pod][flag]
+	s.podTimeIndexCache[pod][flag] = newIndex
+
+	// Trigger an update in pod resources
+	if (!ok || prevIndex != newIndex) && flag == events.PodResources {
+		s.podListenersLock.RLock()
+		if listeners, ok := s.podListeners[events.PodResources]; ok {
+			for _, listener := range listeners {
+				go listener(flagValue)
+			}
+		}
+		s.podListenersLock.RUnlock()
+	}
+}

services/apatelet/store/set_flag.go

@@ -42,7 +42,7 @@ func (s *store) SetPodFlags(label string, flags Flags) {
 	for flag, val := range flags {
 		if listeners, ok := s.podListeners[flag]; ok {
 			for _, listener := range listeners {
-				listener(val)
+				go listener(val)
 			}
 		}
 	}

services/apatelet/store/store_test.go

@@ -515,6 +515,8 @@ func TestAddPodListener(t *testing.T) {
 	st.SetPodFlags("a/b", Flags{events.PodCreatePodResponse: scenario.ResponseTimeout})
 	st.SetPodFlags("b/c", Flags{events.PodUpdatePodResponse: scenario.ResponseError})
 
+	time.Sleep(100 * time.Millisecond) // Sleep a bit as the listeners are executed in goroutines
+
 	// Retrieve unset flag and verify default value and err
 	val, err := st.GetPodFlag(createPodWithLabel("a", "b"), events.PodCreatePodResponse)
 	assert.NoError(t, err)
@@ -585,7 +587,7 @@ func TestGetPodTimeFlag(t *testing.T) {
 	flag, ok = st.getPodTimeFlag(pod, 11, "a/b")
 	assert.False(t, ok)
 	assert.Nil(t, flag)
-	assert.Equal(t, 1, st.podTimeIndexCache[pod][11])
+	assert.Equal(t, 0, st.podTimeIndexCache[pod][11])
 
 	// After 1 more second we expect tf2 to become active
 	time.Sleep(1 * time.Second)
@@ -615,7 +617,7 @@ func TestGetPodTimeFlag(t *testing.T) {
 	assert.Equal(t, 1, st.podTimeIndexCache[pod][11])
 }
 
-func TestScenarioMoreImportantThanPodTime(t *testing.T) {
+func TestPodTimeMoreImportantThanScenario(t *testing.T) {
 	t.Parallel()
 
 	newStore := NewStore()
@@ -635,7 +637,7 @@ func TestScenarioMoreImportantThanPodTime(t *testing.T) {
 
 	flag, err := st.GetPodFlag(createPodWithLabel("a", "b"), 42)
 	assert.NoError(t, err)
-	assert.Equal(t, "k9s", flag)
+	assert.Equal(t, "k8s", flag)
 }
 
 func TestPodTimeMoreImportantThanDefault(t *testing.T) {