TickInfo in LoadDecision

api/aperture/policy/sync/v1/concurrency_limiter.proto

@@ -4,6 +4,7 @@ package aperture.policy.sync.v1;
 
 import "aperture/policy/language/v1/policy.proto";
 import "aperture/policy/sync/v1/common_attributes.proto";
+import "aperture/policy/sync/v1/tick.proto";
 
 message ConcurrencyLimiterWrapper {
   // CommonAttributes
@@ -29,6 +30,7 @@ message TokensDecisionWrapper {
 message LoadDecision {
   double load_multiplier = 1;
   bool pass_through = 2;
+  TickInfo tick_info = 3;
 }
 
 message TokensDecision {

api/aperture/policy/sync/v1/tick.proto

@@ -0,0 +1,13 @@
+syntax = "proto3";
+
+package aperture.policy.sync.v1;
+
+import "google/protobuf/duration.proto";
+import "google/protobuf/timestamp.proto";
+
+message TickInfo {
+  google.protobuf.Timestamp timestamp = 1;
+  google.protobuf.Timestamp next_timestamp = 2;
+  int64 tick = 3;
+  google.protobuf.Duration interval = 4;
+}

pkg/policies/controlplane/components/actuators/concurrency/load-actuator.go

@@ -3,7 +3,6 @@ package concurrency
 import (
 	"context"
 	"path"
-	"sync"
 
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.uber.org/fx"
@@ -24,7 +23,6 @@ import (
 
 // LoadActuator struct.
 type LoadActuator struct {
-	dryRunLock        sync.RWMutex
 	policyReadAPI     iface.Policy
 	decisionWriter    *etcdwriter.Writer
 	loadActuatorProto *policylangv1.LoadActuator
@@ -98,7 +96,7 @@ func (la *LoadActuator) Execute(inPortReadings runtime.PortToValue, tickInfo run
 				} else {
 					lmValue = lmReading.Value()
 				}
-				return nil, la.publishDecision(lmValue, false)
+				return nil, la.publishDecision(tickInfo, lmValue, false)
 			} else {
 				logger.Sample(zerolog.Often).Info().Msg("Invalid load multiplier data")
 			}
@@ -108,13 +106,11 @@ func (la *LoadActuator) Execute(inPortReadings runtime.PortToValue, tickInfo run
 	} else {
 		logger.Sample(zerolog.Often).Info().Msg("load_multiplier port not found")
 	}
-	return nil, la.publishDefaultDecision()
+	return nil, la.publishDefaultDecision(tickInfo)
 }
 
 // DynamicConfigUpdate finds the dynamic config and syncs the decision to agent.
 func (la *LoadActuator) DynamicConfigUpdate(event notifiers.Event, unmarshaller config.Unmarshaller) {
-	la.dryRunLock.Lock()
-	defer la.dryRunLock.Unlock()
 	logger := la.policyReadAPI.GetStatusRegistry().GetLogger()
 	key := la.loadActuatorProto.GetDynamicConfigKey()
 	// read dynamic config
@@ -138,13 +134,11 @@ func (la *LoadActuator) setConfig(config *policylangv1.LoadActuator_DynamicConfi
 	}
 }
 
-func (la *LoadActuator) publishDefaultDecision() error {
-	return la.publishDecision(1.0, true)
+func (la *LoadActuator) publishDefaultDecision(tickInfo runtime.TickInfo) error {
+	return la.publishDecision(tickInfo, 1.0, true)
 }
 
-func (la *LoadActuator) publishDecision(loadMultiplier float64, passThrough bool) error {
-	la.dryRunLock.RLock()
-	defer la.dryRunLock.RUnlock()
+func (la *LoadActuator) publishDecision(tickInfo runtime.TickInfo, loadMultiplier float64, passThrough bool) error {
 	if la.dryRun {
 		passThrough = true
 	}
@@ -153,6 +147,7 @@ func (la *LoadActuator) publishDecision(loadMultiplier float64, passThrough bool
 	decision := &policysyncv1.LoadDecision{
 		LoadMultiplier: loadMultiplier,
 		PassThrough:    passThrough,
+		TickInfo:       tickInfo.Serialize(),
 	}
 	// Publish decision
 	logger.Sample(zerolog.Often).Debug().Float64("loadMultiplier", loadMultiplier).Bool("passThrough", passThrough).Msg("Publish load decision")

pkg/policies/controlplane/runtime/tick-info.go

@@ -1,6 +1,12 @@
 package runtime
 
-import "time"
+import (
+	"time"
+
+	policysyncv1 "github.com/fluxninja/aperture/api/gen/proto/go/aperture/policy/sync/v1"
+	"google.golang.org/protobuf/types/known/durationpb"
+	"google.golang.org/protobuf/types/known/timestamppb"
+)
 
 var _ TickInfo = (*tickInfo)(nil)
 
@@ -10,6 +16,7 @@ type TickInfo interface {
 	NextTimestamp() time.Time
 	Tick() int
 	Interval() time.Duration
+	Serialize() *policysyncv1.TickInfo
 }
 
 type tickInfo struct {
@@ -48,3 +55,13 @@ func (tickInfo *tickInfo) Tick() int {
 func (tickInfo *tickInfo) Interval() time.Duration {
 	return tickInfo.interval
 }
+
+// Serialize returns the proto serialized version of the tickInfo.
+func (tickInfo *tickInfo) Serialize() *policysyncv1.TickInfo {
+	return &policysyncv1.TickInfo{
+		Timestamp:     timestamppb.New(tickInfo.timestamp),
+		NextTimestamp: timestamppb.New(tickInfo.nextTimestamp),
+		Tick:          (int64)(tickInfo.tick),
+		Interval:      durationpb.New(tickInfo.interval),
+	}
+}