diff --git a/proto/policy_action.proto b/proto/policy_action.proto
index 17dd721196c0ed407849aa23099477ea34d39ddd..d547e97790cf2cafe2e4cb9ec8c8b06d8a2e0404 100644
--- a/proto/policy_action.proto
+++ b/proto/policy_action.proto
@@ -26,6 +26,8 @@ enum PolicyRuleActionEnum {
POLICYRULE_ACTION_SET_DEVICE_STATUS = 1;
POLICYRULE_ACTION_ADD_SERVICE_CONFIGRULE = 2;
POLICYRULE_ACTION_ADD_SERVICE_CONSTRAINT = 3;
+ POLICY_RULE_ACTION_CALL_SERVICE_RPC = 4;
+ POLICY_RULE_ACTION_RECALCULATE_PATH = 5;
}
// Action configuration
@@ -37,4 +39,4 @@ message PolicyRuleActionConfig {
// message PolicyRuleAction {
// PolicyRuleActionEnum action = 1;
// repeated string parameters = 2;
-// }
\ No newline at end of file
+// }
diff --git a/src/policy/src/main/java/eu/teraflow/policy/PolicyRuleConditionValidator.java b/src/policy/src/main/java/eu/teraflow/policy/PolicyRuleConditionValidator.java
index 141821958eca02925bdebebdd8a34f46a5844a8a..a1556ffbd715e0065681dcfa6001dd041ff89ef0 100644
--- a/src/policy/src/main/java/eu/teraflow/policy/PolicyRuleConditionValidator.java
+++ b/src/policy/src/main/java/eu/teraflow/policy/PolicyRuleConditionValidator.java
@@ -21,6 +21,7 @@ import eu.teraflow.policy.context.model.Device;
import eu.teraflow.policy.context.model.EndPointId;
import eu.teraflow.policy.context.model.Service;
import eu.teraflow.policy.context.model.ServiceId;
+import eu.teraflow.policy.model.PolicyRuleService;
import io.smallrye.mutiny.Uni;
import java.util.ArrayList;
import java.util.List;
@@ -42,73 +43,44 @@ public class PolicyRuleConditionValidator {
this.contextService = contextService;
}
- public Uni<Boolean> validateDeviceId(String deviceId) {
- final var isDeviceIdValid = isDeviceIdValid(deviceId);
-
- isDeviceIdValid
- .subscribe()
- .with(
- deviceIdIdBooleanValue -> {
- if (Boolean.FALSE.equals(deviceIdIdBooleanValue)) {
- LOGGER.errorf(INVALID_MESSAGE, deviceId);
- }
- LOGGER.infof(VALID_MESSAGE, deviceId);
- });
-
- return isDeviceIdValid;
- }
-
- public Uni<Boolean> validateServiceId(ServiceId serviceId) {
- final var isServiceIdValid = isServiceIdValid(serviceId);
-
- isServiceIdValid
- .subscribe()
- .with(
- serviceIdBooleanValue -> {
- if (Boolean.FALSE.equals(serviceIdBooleanValue)) {
- LOGGER.errorf(INVALID_MESSAGE, serviceId);
- }
- });
-
- return isServiceIdValid;
- }
-
- public Uni<Boolean> validateUpdatedPolicyRuleId(String updatedPolicyRuleId) {
- final var isUpdatedPolicyRuleIdValid = isUpdatedPolicyRuleIdValid(updatedPolicyRuleId);
-
- isUpdatedPolicyRuleIdValid
- .subscribe()
- .with(
- serviceIdBooleanValue -> {
- if (Boolean.FALSE.equals(serviceIdBooleanValue)) {
- LOGGER.errorf(INVALID_MESSAGE, updatedPolicyRuleId);
- }
- });
-
- return isUpdatedPolicyRuleIdValid;
- }
-
- private Uni<Boolean> isDeviceIdValid(String deviceId) {
+ public Uni<Boolean> isDeviceIdValid(String deviceId) {
return contextService
.getDevice(deviceId)
+ .onFailure()
+ .recoverWithItem((Device) null)
.onItem()
.transform(device -> checkIfDeviceIdExists(device, deviceId));
}
private boolean checkIfDeviceIdExists(Device device, String deviceId) {
- final var deviceDeviceId = device.getDeviceId();
+ if (device == null) {
+ return false;
+ }
+ final var deviceDeviceId = device.getDeviceId();
return deviceDeviceId.equals(deviceId);
}
- private Uni<Boolean> isServiceIdValid(ServiceId serviceId) {
+ public Uni<Boolean> isServiceIdValid(ServiceId serviceId, List<String> deviceIds) {
return contextService
.getService(serviceId)
+ .onFailure()
+ .recoverWithItem((Service) null)
.onItem()
- .transform(service -> checkIfServiceIdExists(service, serviceId));
+ .transform(service -> checkIfServiceIsValid(service, serviceId, deviceIds));
+ }
+
+ private boolean checkIfServiceIsValid(
+ Service service, ServiceId serviceId, List<String> deviceIds) {
+ return (checkIfServiceIdExists(service, serviceId)
+ && checkIfServicesDeviceIdsExist(service, deviceIds));
}
private boolean checkIfServiceIdExists(Service service, ServiceId serviceId) {
+ if (service == null) {
+ return false;
+ }
+
final var serviceServiceIdServiceId = service.getServiceId();
final var serviceServiceIdContextId = serviceServiceIdServiceId.getContextId();
final var serviceServiceIdId = serviceServiceIdServiceId.getId();
@@ -117,14 +89,11 @@ public class PolicyRuleConditionValidator {
&& serviceServiceIdId.equals(serviceId.getId());
}
- public Uni<Boolean> isServicesDeviceIdsValid(ServiceId serviceId, List<String> deviceIds) {
- return contextService
- .getService(serviceId)
- .onItem()
- .transform(service -> checkIfServicesDeviceIdsExist(service, deviceIds));
- }
-
private boolean checkIfServicesDeviceIdsExist(Service service, List<String> deviceIds) {
+ if (deviceIds.isEmpty()) {
+ return true;
+ }
+
List<String> serviceDeviceIds = new ArrayList<>();
for (EndPointId serviceEndPointId : service.getServiceEndPointIds()) {
serviceDeviceIds.add(serviceEndPointId.getDeviceId());
@@ -133,7 +102,7 @@ public class PolicyRuleConditionValidator {
return deviceIds.containsAll(serviceDeviceIds);
}
- private Uni<Boolean> isUpdatedPolicyRuleIdValid(String updatedPolicyRuleId) {
+ public Uni<Boolean> isUpdatedPolicyRuleIdValid(String updatedPolicyRuleId) {
return contextService
.getPolicyRule(updatedPolicyRuleId)
.onItem()
@@ -146,4 +115,25 @@ public class PolicyRuleConditionValidator {
.ifNull()
.continueWith(false);
}
+
+ public Uni<Boolean> isPolicyRuleServiceValid(String updatedPolicyRuleId, ServiceId serviceId) {
+ return contextService
+ .getPolicyRule(updatedPolicyRuleId)
+ .onItem()
+ .ifNotNull()
+ .transform(
+ policyRule -> {
+ var policyRuleService =
+ (PolicyRuleService) policyRule.getPolicyRuleType().getPolicyRuleType();
+ if (policyRuleService.getServiceId().getId().equals(serviceId.getId())) {
+ return true;
+ }
+ return false;
+ })
+ .onItem()
+ .ifNull()
+ .continueWith(false)
+ .onFailure()
+ .recoverWithItem(false);
+ }
}
diff --git a/src/policy/src/main/java/eu/teraflow/policy/PolicyServiceImpl.java b/src/policy/src/main/java/eu/teraflow/policy/PolicyServiceImpl.java
index fba43c18a573115d279a251f4806834fdefa2129..8462e0b9d7ec875004cea0a1b2836dc92d3b6f15 100644
--- a/src/policy/src/main/java/eu/teraflow/policy/PolicyServiceImpl.java
+++ b/src/policy/src/main/java/eu/teraflow/policy/PolicyServiceImpl.java
@@ -44,11 +44,10 @@ import eu.teraflow.policy.monitoring.model.AlarmDescriptor;
import eu.teraflow.policy.monitoring.model.AlarmResponse;
import eu.teraflow.policy.monitoring.model.AlarmSubscription;
import eu.teraflow.policy.monitoring.model.KpiValueRange;
-import eu.teraflow.policy.monitoring.model.MonitorKpiRequest;
import eu.teraflow.policy.service.ServiceService;
import io.smallrye.mutiny.Multi;
import io.smallrye.mutiny.Uni;
-import java.time.Duration;
+import io.smallrye.mutiny.groups.UniJoin;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Arrays;
@@ -56,7 +55,6 @@ import java.util.HashMap;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.stream.Collectors;
import javax.enterprise.context.ApplicationScoped;
import javax.inject.Inject;
import org.jboss.logging.Logger;
@@ -71,6 +69,9 @@ public class PolicyServiceImpl implements PolicyService {
private static final int ACCEPTABLE_NUMBER_OF_ALARMS = 3;
private static final int MONITORING_WINDOW_IN_SECONDS = 5;
private static final int SAMPLING_RATE_PER_SECOND = 1;
+ // TODO: Find a better way to disregard alarms while reconfiguring path
+ // Temporary solution for not calling the same rpc more than it's needed
+ private static int noAlarms = 0;
private static final PolicyRuleState INSERTED_POLICYRULE_STATE =
new PolicyRuleState(
@@ -160,94 +161,221 @@ public class PolicyServiceImpl implements PolicyService {
final var policyRuleBasic = policyRuleService.getPolicyRuleBasic();
if (!policyRuleBasic.areArgumentsValid()) {
LOGGER.error(policyRuleService.getExeceptionMessage());
- setPolicyRuleServiceToContext(
- policyRuleService,
+ final var policyRuleState =
new PolicyRuleState(
- PolicyRuleStateEnum.POLICY_FAILED, policyRuleBasic.getExeceptionMessage()));
- return Uni.createFrom().item(policyRuleBasic.getPolicyRuleState());
+ PolicyRuleStateEnum.POLICY_FAILED, policyRuleBasic.getExeceptionMessage());
+ return Uni.createFrom().item(policyRuleState);
}
- policyRuleBasic.setPolicyRuleState(INSERTED_POLICYRULE_STATE);
- policyRuleService.setPolicyRuleBasic(policyRuleBasic);
- final var policyRuleTypeService = new PolicyRuleTypeService(policyRuleService);
- final var policyRule = new PolicyRule(policyRuleTypeService);
+ final var serviceId = policyRuleService.getServiceId();
+ final var deviceIds = policyRuleService.getDeviceIds();
+ final var isServiceValid = policyRuleConditionValidator.isServiceIdValid(serviceId, deviceIds);
+
+ return isServiceValid
+ .onItem()
+ .transform(
+ isService -> {
+ if (!isService) {
+ var policyRuleState =
+ new PolicyRuleState(
+ PolicyRuleStateEnum.POLICY_FAILED,
+ String.format(INVALID_MESSAGE, serviceId));
+
+ return policyRuleState;
+ }
- contextService
- .setPolicyRule(policyRule)
- .subscribe()
- .with(id -> validateService(policyRuleService));
- return Uni.createFrom().item(policyRuleBasic.getPolicyRuleState());
+ final var policyRuleTypeService = new PolicyRuleTypeService(policyRuleService);
+ final var policyRule = new PolicyRule(policyRuleTypeService);
+ final var alarmDescriptorList = createAlarmDescriptorList(policyRule);
+
+ if (alarmDescriptorList.isEmpty()) {
+ var policyRuleState =
+ new PolicyRuleState(
+ PolicyRuleStateEnum.POLICY_FAILED,
+ String.format(
+ "Invalid PolicyRuleConditions in PolicyRule with ID: %s",
+ policyRuleBasic.getPolicyRuleId()));
+ return policyRuleState;
+ }
+ contextService.setPolicyRule(policyRule).subscribe().with(x -> {});
+ setPolicyRuleServiceToContext(policyRuleService, VALIDATED_POLICYRULE_STATE);
+ noAlarms = 0;
+
+ // Create an alarmIds list that contains the promised ids returned from setKpiAlarm
+ List<Uni<String>> alarmIds = new ArrayList<Uni<String>>();
+ for (AlarmDescriptor alarmDescriptor : alarmDescriptorList) {
+ LOGGER.infof("alarmDescriptor:");
+ LOGGER.infof(alarmDescriptor.toString());
+ alarmIds.add(monitoringService.setKpiAlarm(alarmDescriptor));
+ }
+
+ // Transform the alarmIds into promised alarms returned from the
+ // getAlarmResponseStream
+ List<Multi<AlarmResponse>> alarmResponseStreamList = new ArrayList<>();
+ for (Uni<String> alarmId : alarmIds) {
+ alarmResponseStreamList.add(
+ alarmId
+ .onItem()
+ .transformToMulti(
+ id -> {
+ alarmPolicyRuleServiceMap.put(id, policyRuleService);
+
+ // TODO: Create infinite subscription
+ var alarmSubscription = new AlarmSubscription(id, 259200, 5000);
+ return monitoringService.getAlarmResponseStream(alarmSubscription);
+ }));
+ }
+
+ // Merge the promised alarms into one stream (Multi Object)
+ final var multi = Multi.createBy().merging().streams(alarmResponseStreamList);
+ setPolicyRuleServiceToContext(policyRuleService, PROVISIONED_POLICYRULE_STATE);
+
+ monitorAlarmResponseForService(multi);
+
+ // TODO: Resubscribe to the stream, if it has ended
+
+ // TODO: Redesign evaluation of action
+ // evaluateAction(policyRule, alarmDescriptorList, multi);
+
+ return VALIDATED_POLICYRULE_STATE;
+ });
}
@Override
- public Uni<PolicyRuleState> updatePolicyService(PolicyRuleService policyRuleService) {
- LOGGER.infof("Received %s", policyRuleService);
+ public Uni<PolicyRuleState> addPolicyDevice(PolicyRuleDevice policyRuleDevice) {
+ LOGGER.infof("Received %s", policyRuleDevice);
- if (!policyRuleService.areArgumentsValid()) {
- LOGGER.error(policyRuleService.getExeceptionMessage());
+ if (!policyRuleDevice.areArgumentsValid()) {
+ LOGGER.error(policyRuleDevice.getExeceptionMessage());
final var policyRuleState =
new PolicyRuleState(
- PolicyRuleStateEnum.POLICY_FAILED, policyRuleService.getExeceptionMessage());
+ PolicyRuleStateEnum.POLICY_FAILED, policyRuleDevice.getExeceptionMessage());
return Uni.createFrom().item(policyRuleState);
}
- final var policyRuleBasic = policyRuleService.getPolicyRuleBasic();
+ final var policyRuleBasic = policyRuleDevice.getPolicyRuleBasic();
if (!policyRuleBasic.areArgumentsValid()) {
- LOGGER.error(policyRuleService.getExeceptionMessage());
- setPolicyRuleServiceToContext(
- policyRuleService,
+ LOGGER.error(policyRuleDevice.getExeceptionMessage());
+ final var policyRuleState =
new PolicyRuleState(
- PolicyRuleStateEnum.POLICY_FAILED, policyRuleBasic.getExeceptionMessage()));
- return Uni.createFrom().item(policyRuleBasic.getPolicyRuleState());
+ PolicyRuleStateEnum.POLICY_FAILED, policyRuleBasic.getExeceptionMessage());
+ return Uni.createFrom().item(policyRuleState);
}
- policyRuleBasic.setPolicyRuleState(UPDATED_POLICYRULE_STATE);
- policyRuleService.setPolicyRuleBasic(policyRuleBasic);
- final var policyRuleTypeService = new PolicyRuleTypeService(policyRuleService);
- final var policyRule = new PolicyRule(policyRuleTypeService);
+ final var deviceIds = policyRuleDevice.getDeviceIds();
+ final var areDevicesValid = returnInvalidDeviceIds(deviceIds);
+
+ return areDevicesValid
+ .onItem()
+ .transform(
+ areDevices -> {
+ if (areDevices.contains(false)) {
+ var policyRuleState =
+ new PolicyRuleState(
+ PolicyRuleStateEnum.POLICY_FAILED,
+ String.format(
+ INVALID_MESSAGE,
+ policyRuleDevice.getPolicyRuleBasic().getPolicyRuleId()));
+
+ return policyRuleState;
+ }
- contextService
- .setPolicyRule(policyRule)
- .subscribe()
- .with(id -> validateUpdatedPolicyService(policyRuleService));
+ final var policyRuleTypeDevice = new PolicyRuleTypeDevice(policyRuleDevice);
+ final var policyRule = new PolicyRule(policyRuleTypeDevice);
+
+ final var alarmDescriptorList = createAlarmDescriptorList(policyRule);
+ if (alarmDescriptorList.isEmpty()) {
+ var policyRuleState =
+ new PolicyRuleState(
+ PolicyRuleStateEnum.POLICY_FAILED,
+ String.format(
+ "Invalid PolicyRuleConditions in PolicyRule with ID: %s",
+ policyRuleBasic.getPolicyRuleId()));
+ return policyRuleState;
+ }
+
+ contextService.setPolicyRule(policyRule).subscribe().with(x -> {});
+ setPolicyRuleDeviceToContext(policyRuleDevice, VALIDATED_POLICYRULE_STATE);
+ noAlarms = 0;
+
+ List<Uni<String>> alarmIds = new ArrayList<Uni<String>>();
+ for (AlarmDescriptor alarmDescriptor : alarmDescriptorList) {
+ LOGGER.infof("alarmDescriptor:");
+ LOGGER.infof(alarmDescriptor.toString());
+ alarmIds.add(monitoringService.setKpiAlarm(alarmDescriptor));
+ }
+
+ // Transform the alarmIds into promised alarms returned from the
+ // getAlarmResponseStream
+ List<Multi<AlarmResponse>> alarmResponseStreamList = new ArrayList<>();
+ for (Uni<String> alarmId : alarmIds) {
+ alarmResponseStreamList.add(
+ alarmId
+ .onItem()
+ .transformToMulti(
+ id -> {
+ alarmPolicyRuleDeviceMap.put(id, policyRuleDevice);
+
+ // TODO: Create infinite subscription
+ var alarmSubscription = new AlarmSubscription(id, 259200, 5000);
+ return monitoringService.getAlarmResponseStream(alarmSubscription);
+ }));
+ }
+
+ // Merge the promised alarms into one stream (Multi Object)
+ final var multi = Multi.createBy().merging().streams(alarmResponseStreamList);
+ setPolicyRuleDeviceToContext(policyRuleDevice, PROVISIONED_POLICYRULE_STATE);
- return Uni.createFrom().item(policyRuleBasic.getPolicyRuleState());
+ monitorAlarmResponseForDevice(multi);
+
+ // TODO: Resubscribe to the stream, if it has ended
+
+ // TODO: Redesign evaluation of action
+ // evaluateAction(policyRule, alarmDescriptorList, multi);
+
+ return VALIDATED_POLICYRULE_STATE;
+ });
}
@Override
- public Uni<PolicyRuleState> addPolicyDevice(PolicyRuleDevice policyRuleDevice) {
- LOGGER.infof("Received %s", policyRuleDevice);
+ public Uni<PolicyRuleState> updatePolicyService(PolicyRuleService policyRuleService) {
+ LOGGER.infof("Received %s", policyRuleService);
- if (!policyRuleDevice.areArgumentsValid()) {
- LOGGER.error(policyRuleDevice.getExeceptionMessage());
+ if (!policyRuleService.areArgumentsValid()) {
+ LOGGER.error(policyRuleService.getExeceptionMessage());
final var policyRuleState =
new PolicyRuleState(
- PolicyRuleStateEnum.POLICY_FAILED, policyRuleDevice.getExeceptionMessage());
+ PolicyRuleStateEnum.POLICY_FAILED, policyRuleService.getExeceptionMessage());
return Uni.createFrom().item(policyRuleState);
}
- final var policyRuleBasic = policyRuleDevice.getPolicyRuleBasic();
+ final var policyRuleBasic = policyRuleService.getPolicyRuleBasic();
if (!policyRuleBasic.areArgumentsValid()) {
- LOGGER.error(policyRuleDevice.getExeceptionMessage());
- setPolicyRuleDeviceToContext(
- policyRuleDevice,
+ LOGGER.error(policyRuleService.getExeceptionMessage());
+ final var policyRuleState =
new PolicyRuleState(
- PolicyRuleStateEnum.POLICY_FAILED, policyRuleBasic.getExeceptionMessage()));
- return Uni.createFrom().item(policyRuleBasic.getPolicyRuleState());
+ PolicyRuleStateEnum.POLICY_FAILED, policyRuleBasic.getExeceptionMessage());
+ return Uni.createFrom().item(policyRuleState);
}
- policyRuleBasic.setPolicyRuleState(INSERTED_POLICYRULE_STATE);
- policyRuleDevice.setPolicyRuleBasic(policyRuleBasic);
- final var policyRuleTypeDevice = new PolicyRuleTypeDevice(policyRuleDevice);
- final var policyRule = new PolicyRule(policyRuleTypeDevice);
+ final var serviceId = policyRuleService.getServiceId();
+ final var policyRuleId = policyRuleBasic.getPolicyRuleId();
+ final var isPolicyRuleServiceValid =
+ policyRuleConditionValidator.isPolicyRuleServiceValid(policyRuleId, serviceId);
+
+ return isPolicyRuleServiceValid
+ .onItem()
+ .transform(
+ isPolicyRuleService -> {
+ if (!isPolicyRuleService) {
+ return new PolicyRuleState(
+ PolicyRuleStateEnum.POLICY_FAILED, String.format(INVALID_MESSAGE, serviceId));
+ }
- contextService
- .setPolicyRule(policyRule)
- .subscribe()
- .with(id -> validateDevice(policyRuleDevice));
- return Uni.createFrom().item(policyRuleBasic.getPolicyRuleState());
+ return VALIDATED_POLICYRULE_STATE;
+ });
}
@Override
@@ -265,153 +393,154 @@ public class PolicyServiceImpl implements PolicyService {
final var policyRuleBasic = policyRuleDevice.getPolicyRuleBasic();
if (!policyRuleBasic.areArgumentsValid()) {
- LOGGER.error(policyRuleDevice.getExeceptionMessage());
- setPolicyRuleDeviceToContext(
- policyRuleDevice,
+ final var policyRuleState =
new PolicyRuleState(
- PolicyRuleStateEnum.POLICY_FAILED, policyRuleBasic.getExeceptionMessage()));
- return Uni.createFrom().item(policyRuleBasic.getPolicyRuleState());
+ PolicyRuleStateEnum.POLICY_FAILED, policyRuleBasic.getExeceptionMessage());
+ return Uni.createFrom().item(policyRuleState);
}
- policyRuleBasic.setPolicyRuleState(UPDATED_POLICYRULE_STATE);
- policyRuleDevice.setPolicyRuleBasic(policyRuleBasic);
- final var policyRuleTypeDevice = new PolicyRuleTypeDevice(policyRuleDevice);
- final var policyRule = new PolicyRule(policyRuleTypeDevice);
+ final var policyRuleId = policyRuleBasic.getPolicyRuleId();
+ final var isPolicyRuleValid =
+ policyRuleConditionValidator.isUpdatedPolicyRuleIdValid(policyRuleId);
- contextService
- .setPolicyRule(policyRule)
- .subscribe()
- .with(id -> validateUpdatedPolicyDevice(policyRuleDevice));
+ return isPolicyRuleValid
+ .onItem()
+ .transform(
+ isPolicyRuleService -> {
+ if (!isPolicyRuleService) {
+ return new PolicyRuleState(
+ PolicyRuleStateEnum.POLICY_FAILED,
+ String.format(INVALID_MESSAGE, policyRuleId));
+ }
- return Uni.createFrom().item(policyRuleBasic.getPolicyRuleState());
+ return VALIDATED_POLICYRULE_STATE;
+ });
}
@Override
public Uni<PolicyRuleState> deletePolicy(String policyRuleId) {
LOGGER.infof("Received %s", policyRuleId);
- PolicyRule policyRule = contextService.getPolicyRule(policyRuleId).await().indefinitely();
-
- final var policyRuleBasic = policyRule.getPolicyRuleType().getPolicyRuleBasic();
- List<PolicyRuleCondition> policyRuleConditions = policyRuleBasic.getPolicyRuleConditions();
-
- for (PolicyRuleCondition policy : policyRuleConditions) {
- var empty = monitoringService.deleteKpi(policy.getKpiId());
- empty
- .subscribe()
- .with(emptyMessage -> LOGGER.infof("Kpi [%s] has been deleted.\n", policyRuleId));
- }
-
- var empty = contextService.removePolicyRule(policyRuleId);
- empty
- .subscribe()
- .with(emptyMessage -> LOGGER.infof("Policy [%s] has been removed.\n", policyRuleId));
-
- setPolicyRuleToContext(policyRule, REMOVED_POLICYRULE_STATE);
- return Uni.createFrom().item(policyRuleBasic.getPolicyRuleState());
+ final var getPolicyRule = contextService.getPolicyRule(policyRuleId);
+
+ return getPolicyRule
+ .onItem()
+ .transform(
+ policyRule -> {
+ var policyRuleBasic = policyRule.getPolicyRuleType().getPolicyRuleBasic();
+
+ policyRule
+ .getPolicyRuleType()
+ .getPolicyRuleBasic()
+ .setPolicyRuleState(REMOVED_POLICYRULE_STATE);
+
+ contextService
+ .setPolicyRule(policyRule)
+ .subscribe()
+ .with(
+ tmp ->
+ LOGGER.infof(
+ "DeletePolicy with id: " + VALID_MESSAGE,
+ policyRuleBasic.getPolicyRuleId()));
+
+ return policyRuleBasic.getPolicyRuleState();
+ });
}
- private void monitorKpi(List<AlarmDescriptor> alarmDescriptorList) {
-
- for (AlarmDescriptor alarmDescriptor : alarmDescriptorList) {
- var monitorKpiRequest =
- new MonitorKpiRequest(
- alarmDescriptor.getKpiId(), MONITORING_WINDOW_IN_SECONDS, SAMPLING_RATE_PER_SECOND);
- monitoringService
- .monitorKpi(monitorKpiRequest)
- .subscribe()
- .with(
- emptyMessage ->
- LOGGER.infof(
- "Kpi [%s] has started to be monitored.\n", alarmDescriptor.getKpiId()));
+ private Uni<List<Boolean>> returnInvalidDeviceIds(List<String> deviceIds) {
+ UniJoin.Builder<Boolean> builder = Uni.join().builder();
+ for (String deviceId : deviceIds) {
+ final var validatedDeviceId = policyRuleConditionValidator.isDeviceIdValid(deviceId);
+ builder.add(validatedDeviceId);
}
+ return builder.joinAll().andFailFast();
}
- private void provisionAlarm(
- PolicyRule policyRule, List<AlarmDescriptor> alarmDescriptorList, Boolean isService) {
+ private List<AlarmDescriptor> createAlarmDescriptorList(PolicyRule policyRule) {
+ final var policyRuleType = policyRule.getPolicyRuleType();
+ final var policyRuleTypeSpecificType = policyRuleType.getPolicyRuleType();
- List<AlarmSubscription> alarmSubscriptionList = new ArrayList<>();
+ List<AlarmDescriptor> alarmDescriptorList = new ArrayList<>();
+ if (policyRuleTypeSpecificType instanceof PolicyRuleService) {
+ final var policyRuleService = (PolicyRuleService) policyRuleTypeSpecificType;
+ final var policyRuleBasic = policyRuleService.getPolicyRuleBasic();
- for (AlarmDescriptor alarmDescriptor : alarmDescriptorList) {
- monitoringService
- .setKpiAlarm(alarmDescriptor)
- .subscribe()
- .with(alarmId -> alarmSubscriptionList.add(new AlarmSubscription(alarmId, 0, 0)));
- }
+ alarmDescriptorList = parsePolicyRuleCondition(policyRuleBasic);
+ if (alarmDescriptorList.isEmpty()) {
+ return List.of();
+ }
+ } else {
+ final var policyRuleDevice = (PolicyRuleDevice) policyRuleTypeSpecificType;
+ final var policyRuleBasic = policyRuleDevice.getPolicyRuleBasic();
- setPolicyRuleToContext(policyRule, PROVISIONED_POLICYRULE_STATE);
+ alarmDescriptorList = parsePolicyRuleCondition(policyRuleBasic);
+ if (alarmDescriptorList.isEmpty()) {
+ return List.of();
+ }
+ for (AlarmDescriptor alarmDescriptor : alarmDescriptorList) {
+ alarmPolicyRuleDeviceMap.put(alarmDescriptor.getAlarmId(), policyRuleDevice);
+ }
+ }
- getAlarmResponseStream(policyRule, alarmDescriptorList, alarmSubscriptionList, isService);
+ return alarmDescriptorList;
}
- private void getAlarmResponseStream(
- PolicyRule policyRule,
- List<AlarmDescriptor> alarmDescriptorList,
- List<AlarmSubscription> alarmSubscriptionList,
- Boolean isService) {
-
- List<Multi<AlarmResponse>> alarmResponseStreamList = new ArrayList<>();
- for (AlarmSubscription alarmSubscription : alarmSubscriptionList) {
- alarmResponseStreamList.add(monitoringService.getAlarmResponseStream(alarmSubscription));
- }
- Multi<AlarmResponse> multi = Multi.createBy().merging().streams(alarmResponseStreamList);
-
+ private void monitorAlarmResponseForService(Multi<AlarmResponse> multi) {
multi
- .select()
- .first()
.subscribe()
.with(
alarmResponse -> {
+ LOGGER.infof("**************************Received Alarm!**************************");
+ LOGGER.infof("alarmResponse:");
LOGGER.info(alarmResponse);
- if (isService) {
- if (!alarmPolicyRuleServiceMap.containsKey(alarmResponse.getAlarmId())) {
- return;
- }
- applyActionService(alarmResponse.getAlarmId());
- } else {
- if (!alarmPolicyRuleDeviceMap.containsKey(alarmResponse.getAlarmId())) {
- return;
- }
- applyActionDevice(alarmResponse.getAlarmId());
- }
+ LOGGER.info(alarmResponse.getAlarmId());
+ applyActionService(alarmResponse.getAlarmId());
});
+ }
- Long count =
- multi
- .collect()
- .with(Collectors.counting())
- .await()
- .atMost(Duration.ofMinutes(POLICY_EVALUATION_TIMEOUT));
-
- if (count > ACCEPTABLE_NUMBER_OF_ALARMS) {
- for (AlarmDescriptor alarmDescriptor : alarmDescriptorList) {
- monitoringService
- .deleteAlarm(alarmDescriptor.getAlarmId())
- .subscribe()
- .with(
- emptyMessage ->
- LOGGER.infof("Alarm [%s] has been deleted.\n", alarmDescriptor.getAlarmId()));
- }
-
- setPolicyRuleToContext(policyRule, INEFFECTIVE_POLICYRULE_STATE);
-
- } else {
- setPolicyRuleToContext(policyRule, EFFECTIVE_POLICYRULE_STATE);
-
- multi
- .subscribe()
- .with(
- alarmResponse -> {
- LOGGER.info(alarmResponse);
- if (isService) {
- applyActionService(alarmResponse.getAlarmId());
- } else {
- applyActionDevice(alarmResponse.getAlarmId());
- }
- });
- }
+ private void monitorAlarmResponseForDevice(Multi<AlarmResponse> multi) {
+ multi
+ .subscribe()
+ .with(
+ alarmResponse -> {
+ LOGGER.infof("**************************Received Alarm!**************************");
+ LOGGER.infof("alarmResponse:");
+ LOGGER.info(alarmResponse);
+ LOGGER.info(alarmResponse.getAlarmId());
+ applyActionDevice(alarmResponse.getAlarmId());
+ });
}
+ // TODO: To be refactored or deprecated
+ // private void evaluateAction(
+ // PolicyRule policyRule,
+ // List<AlarmDescriptor> alarmDescriptorList,
+ // Multi<AlarmResponse> multi) {
+ //
+ // Long count =
+ // multi
+ // .collect()
+ // .with(Collectors.counting())
+ // .await()
+ // .atMost(Duration.ofMinutes(POLICY_EVALUATION_TIMEOUT));
+ //
+ // if (count > ACCEPTABLE_NUMBER_OF_ALARMS) {
+ // for (AlarmDescriptor alarmDescriptor : alarmDescriptorList) {
+ // monitoringService
+ // .deleteAlarm(alarmDescriptor.getAlarmId())
+ // .subscribe()
+ // .with(
+ // emptyMessage ->
+ // LOGGER.infof(
+ // "Alarm [%s] has been deleted as ineffective.\n",
+ // alarmDescriptor.getAlarmId()));
+ // }
+ // setPolicyRuleToContext(policyRule, INEFFECTIVE_POLICYRULE_STATE);
+ // } else {
+ // setPolicyRuleToContext(policyRule, EFFECTIVE_POLICYRULE_STATE);
+ // }
+ // }
+
private void applyActionDevice(String alarmId) {
PolicyRuleDevice policyRuleDevice = alarmPolicyRuleDeviceMap.get(alarmId);
@@ -526,190 +655,52 @@ public class PolicyServiceImpl implements PolicyService {
});
}
- private void applyActionService(String alarmId) {
- PolicyRuleService policyRuleService = alarmPolicyRuleServiceMap.get(alarmId);
- PolicyRuleAction policyRuleAction = policyRuleActionMap.get(alarmId);
-
- setPolicyRuleServiceToContext(policyRuleService, ACTIVE_POLICYRULE_STATE);
-
- switch (policyRuleAction.getPolicyRuleActionEnum()) {
- case POLICY_RULE_ACTION_ADD_SERVICE_CONSTRAINT:
- addServiceConstraint(policyRuleService, policyRuleAction);
- case POLICY_RULE_ACTION_ADD_SERVICE_CONFIGRULE:
- addServiceConfigRule(policyRuleService, policyRuleAction);
- default:
- LOGGER.errorf(INVALID_MESSAGE, policyRuleAction.getPolicyRuleActionEnum());
- return;
- }
- }
-
- private void validateDevice(PolicyRuleDevice policyRuleDevice) {
- final var deviceIds = policyRuleDevice.getDeviceIds();
- final var policyRuleId = policyRuleDevice.getPolicyRuleBasic().getPolicyRuleId();
-
- final var invalidDeviceIds = returnInvalidDeviceIds(deviceIds);
- if (!invalidDeviceIds.isEmpty()) {
- String ids = "";
- for (String id : invalidDeviceIds) {
- ids += " ," + id;
- }
-
- String message =
- String.format(
- "The following devices in PolicyRuleDevice with ID: %s are not valid: %s",
- policyRuleId, ids);
- setPolicyRuleDeviceToContext(
- policyRuleDevice, new PolicyRuleState(PolicyRuleStateEnum.POLICY_FAILED, message));
- return;
- }
-
- createAlarmDescriptorsForDevices(policyRuleDevice);
- }
-
- private void createAlarmDescriptorsForDevices(PolicyRuleDevice policyRuleDevice) {
- final var policyRuleBasic = policyRuleDevice.getPolicyRuleBasic();
-
- List<AlarmDescriptor> alarmDescriptorList =
- parsePolicyRuleCondition(policyRuleDevice.getPolicyRuleBasic());
-
- if (alarmDescriptorList.isEmpty()) {
- String message =
- String.format(
- "The devices of PolicyRuleDevice with ID: %s are not valid",
- policyRuleBasic.getPolicyRuleId());
- setPolicyRuleDeviceToContext(
- policyRuleDevice, new PolicyRuleState(PolicyRuleStateEnum.POLICY_FAILED, message));
- return;
- }
-
- setPolicyRuleDeviceToContext(policyRuleDevice, VALIDATED_POLICYRULE_STATE);
- for (AlarmDescriptor alarmDescriptor : alarmDescriptorList) {
- alarmPolicyRuleDeviceMap.put(alarmDescriptor.getAlarmId(), policyRuleDevice);
- }
-
- final var policyRuleTypeService = new PolicyRuleTypeDevice(policyRuleDevice);
- final var policyRule = new PolicyRule(policyRuleTypeService);
- monitorKpi(alarmDescriptorList);
- provisionAlarm(policyRule, alarmDescriptorList, false);
- return;
- }
-
- private void validateUpdatedPolicyService(PolicyRuleService policyRuleService) {
-
- final var policyRuleBasic = policyRuleService.getPolicyRuleBasic();
- final var isUpdatedPolicyRuleValid =
- policyRuleConditionValidator.validateUpdatedPolicyRuleId(policyRuleBasic.getPolicyRuleId());
-
- isUpdatedPolicyRuleValid
- .subscribe()
- .with(
- policyRuleBoolean -> {
- if (Boolean.FALSE.equals(policyRuleBoolean)) {
-
- String message =
- String.format(
- "The PolicyRule with ID: %s was not found. PolicyUpdateService failed.",
- policyRuleBasic.getPolicyRuleId());
- setPolicyRuleServiceToContext(
- policyRuleService,
- new PolicyRuleState(PolicyRuleStateEnum.POLICY_FAILED, message));
- return;
- }
-
- validateService(policyRuleService);
- });
- }
-
- private void validateUpdatedPolicyDevice(PolicyRuleDevice policyRuleDevice) {
+ private void callRecalculatePathRPC(
+ PolicyRuleService policyRuleService, PolicyRuleAction policyRuleAction) {
- final var policyRuleBasic = policyRuleDevice.getPolicyRuleBasic();
- final var isUpdatedPolicyRuleValid =
- policyRuleConditionValidator.validateUpdatedPolicyRuleId(policyRuleBasic.getPolicyRuleId());
+ final var deserializedServiceUni = contextService.getService(policyRuleService.getServiceId());
- isUpdatedPolicyRuleValid
+ deserializedServiceUni
.subscribe()
.with(
- policyRuleBoolean -> {
- if (Boolean.FALSE.equals(policyRuleBoolean)) {
- String message =
- String.format(
- "PolicyRule with ID: %s was not found. PolicyUpdateDevice failed",
- policyRuleBasic.getPolicyRuleId());
- setPolicyRuleDeviceToContext(
- policyRuleDevice,
- new PolicyRuleState(PolicyRuleStateEnum.POLICY_FAILED, message));
- return;
- }
- validateDevice(policyRuleDevice);
+ deserializedService -> {
+ serviceService
+ .recomputeConnections(deserializedService)
+ .subscribe()
+ .with(
+ x -> {
+ LOGGER.info("called recomputeConnections with:");
+ LOGGER.info(deserializedService);
+ setPolicyRuleServiceToContext(policyRuleService, ENFORCED_POLICYRULE_STATE);
+ });
});
}
- private void validateService(PolicyRuleService policyRuleService) {
- final var serviceId = policyRuleService.getServiceId();
- final var deviceIds = policyRuleService.getDeviceIds();
- final var policyRuleBasic = policyRuleService.getPolicyRuleBasic();
-
- Boolean isServiceIdValid =
- policyRuleConditionValidator.validateServiceId(serviceId).await().indefinitely();
-
- if (!isServiceIdValid) {
- String message =
- String.format(
- "Cannot provision/update a PolicyRule with invalid service ID: %s",
- policyRuleBasic.getPolicyRuleId());
- setPolicyRuleServiceToContext(
- policyRuleService, new PolicyRuleState(PolicyRuleStateEnum.POLICY_FAILED, message));
- return;
- }
-
- Boolean isServicesDeviceIdsValid =
- policyRuleConditionValidator
- .isServicesDeviceIdsValid(serviceId, deviceIds)
- .await()
- .indefinitely();
-
- if (!isServicesDeviceIdsValid) {
-
- String message =
- String.format(
- "Cannot provision/update a PolicyRule with invalid service ID: %s",
- policyRuleBasic.getPolicyRuleId());
- setPolicyRuleServiceToContext(
- policyRuleService, new PolicyRuleState(PolicyRuleStateEnum.POLICY_FAILED, message));
- return;
- }
-
- setPolicyRuleServiceToContext(policyRuleService, VALIDATED_POLICYRULE_STATE);
-
- createAlarmDescriptorsForService(policyRuleService);
- }
-
- private void createAlarmDescriptorsForService(PolicyRuleService policyRuleService) {
- final var policyRuleBasic = policyRuleService.getPolicyRuleBasic();
-
- List<AlarmDescriptor> alarmDescriptorList =
- parsePolicyRuleCondition(policyRuleService.getPolicyRuleBasic());
-
- if (alarmDescriptorList.isEmpty()) {
- String message =
- String.format(
- "Invalid PolicyRuleConditions in PolicyRule with ID: %s",
- policyRuleBasic.getPolicyRuleId());
- setPolicyRuleServiceToContext(
- policyRuleService, new PolicyRuleState(PolicyRuleStateEnum.POLICY_FAILED, message));
- return;
- }
-
- setPolicyRuleServiceToContext(policyRuleService, VALIDATED_POLICYRULE_STATE);
-
- for (AlarmDescriptor alarmDescriptor : alarmDescriptorList) {
- alarmPolicyRuleServiceMap.put(alarmDescriptor.getAlarmId(), policyRuleService);
+ private void applyActionService(String alarmId) {
+ PolicyRuleService policyRuleService = alarmPolicyRuleServiceMap.get(alarmId);
+ PolicyRuleAction policyRuleAction =
+ policyRuleService.getPolicyRuleBasic().getPolicyRuleActions().get(0);
+
+ if (noAlarms == 0) {
+ noAlarms++;
+ setPolicyRuleServiceToContext(policyRuleService, ACTIVE_POLICYRULE_STATE);
+
+ switch (policyRuleAction.getPolicyRuleActionEnum()) {
+ case POLICY_RULE_ACTION_ADD_SERVICE_CONSTRAINT:
+ addServiceConstraint(policyRuleService, policyRuleAction);
+ case POLICY_RULE_ACTION_ADD_SERVICE_CONFIGRULE:
+ addServiceConfigRule(policyRuleService, policyRuleAction);
+ case POLICY_RULE_ACTION_RECALCULATE_PATH:
+ callRecalculatePathRPC(policyRuleService, policyRuleAction);
+ default:
+ LOGGER.errorf(INVALID_MESSAGE, policyRuleAction.getPolicyRuleActionEnum());
+ return;
+ }
+ } else if (noAlarms == 2) {
+ noAlarms = 0;
+ } else {
+ noAlarms++;
}
-
- final var policyRuleTypeService = new PolicyRuleTypeService(policyRuleService);
- final var policyRule = new PolicyRule(policyRuleTypeService);
- provisionAlarm(policyRule, alarmDescriptorList, true);
- return;
}
private List<AlarmDescriptor> parsePolicyRuleCondition(PolicyRuleBasic policyRuleBasic) {
@@ -833,16 +824,16 @@ public class PolicyServiceImpl implements PolicyService {
false);
case POLICY_RULE_CONDITION_NUMERICAL_GREATER_THAN:
- return new KpiValueRange(policyRuleCondition.getKpiValue(), null, false, false, false);
+ return new KpiValueRange(null, policyRuleCondition.getKpiValue(), true, false, false);
case POLICY_RULE_CONDITION_NUMERICAL_GREATER_THAN_EQUAL:
- return new KpiValueRange(policyRuleCondition.getKpiValue(), null, false, true, false);
+ return new KpiValueRange(null, policyRuleCondition.getKpiValue(), true, true, false);
case POLICY_RULE_CONDITION_NUMERICAL_LESS_THAN:
- return new KpiValueRange(null, policyRuleCondition.getKpiValue(), false, false, false);
+ return new KpiValueRange(policyRuleCondition.getKpiValue(), null, true, false, false);
case POLICY_RULE_CONDITION_NUMERICAL_LESS_THAN_EQUAL:
- return new KpiValueRange(null, policyRuleCondition.getKpiValue(), false, false, true);
+ return new KpiValueRange(policyRuleCondition.getKpiValue(), null, true, false, true);
default:
return null;
}
@@ -885,31 +876,6 @@ public class PolicyServiceImpl implements PolicyService {
return new KpiValueRange(kpiMinValue, kpiMaxValue, true, includeMinValue, includeMaxValue);
}
- private List<String> returnInvalidDeviceIds(List<String> deviceIds) {
- var invalidDeviceIds = new ArrayList<String>();
-
- if (!deviceIds.isEmpty()) {
-
- for (String deviceId : deviceIds) {
- final var validatedDeviceId = policyRuleConditionValidator.validateDeviceId(deviceId);
-
- validatedDeviceId
- .subscribe()
- .with(
- deviceIdBoolean -> {
- if (Boolean.FALSE.equals(deviceIdBoolean)) {
- invalidDeviceIds.add(deviceId);
- }
- });
- }
-
- } else {
- LOGGER.warnf("No deviceIds found");
- }
-
- return invalidDeviceIds;
- }
-
private void setPolicyRuleToContext(PolicyRule policyRule, PolicyRuleState policyRuleState) {
final var policyRuleType = policyRule.getPolicyRuleType();
final var policyRuleTypeSpecificType = policyRuleType.getPolicyRuleType();
diff --git a/src/policy/src/main/java/eu/teraflow/policy/context/model/ServiceStatusEnum.java b/src/policy/src/main/java/eu/teraflow/policy/context/model/ServiceStatusEnum.java
index 0a9115acd7cec634d1a4cf6dd096bf44adab141e..353b78eba1ad261d6c65561d1fe3e4f64a914420 100644
--- a/src/policy/src/main/java/eu/teraflow/policy/context/model/ServiceStatusEnum.java
+++ b/src/policy/src/main/java/eu/teraflow/policy/context/model/ServiceStatusEnum.java
@@ -20,5 +20,7 @@ public enum ServiceStatusEnum {
UNDEFINED,
PLANNED,
ACTIVE,
- PENDING_REMOVAL
+ PENDING_REMOVAL,
+ SLA_VIOLATED,
+ UPDATING
}
diff --git a/src/policy/src/main/java/eu/teraflow/policy/model/PolicyRuleActionEnum.java b/src/policy/src/main/java/eu/teraflow/policy/model/PolicyRuleActionEnum.java
index 432c3834c1c2f3417540bb4e62d43a1dfc58a167..971dfbaf29bba86ce0043e2edc6bb1e21f181643 100644
--- a/src/policy/src/main/java/eu/teraflow/policy/model/PolicyRuleActionEnum.java
+++ b/src/policy/src/main/java/eu/teraflow/policy/model/PolicyRuleActionEnum.java
@@ -20,5 +20,8 @@ public enum PolicyRuleActionEnum {
POLICY_RULE_ACTION_NO_ACTION,
POLICY_RULE_ACTION_SET_DEVICE_STATUS,
POLICY_RULE_ACTION_ADD_SERVICE_CONFIGRULE,
- POLICY_RULE_ACTION_ADD_SERVICE_CONSTRAINT
+ POLICY_RULE_ACTION_ADD_SERVICE_CONSTRAINT,
+ POLICY_RULE_ACTION_CALL_SERVICE_RPC,
+ // This is temporary
+ POLICY_RULE_ACTION_RECALCULATE_PATH
}
diff --git a/src/policy/src/main/java/eu/teraflow/policy/monitoring/model/AlarmSubscription.java b/src/policy/src/main/java/eu/teraflow/policy/monitoring/model/AlarmSubscription.java
index 532eed60868dfbd5fe1996c781964d3d49a368af..7b99c20456987b6c56a0af00821fb5ef33954ada 100644
--- a/src/policy/src/main/java/eu/teraflow/policy/monitoring/model/AlarmSubscription.java
+++ b/src/policy/src/main/java/eu/teraflow/policy/monitoring/model/AlarmSubscription.java
@@ -22,6 +22,7 @@ public class AlarmSubscription {
private final float subscriptionTimeoutS;
private final float subscriptionFrequencyMs;
+ // TODO: Refactor the AlarmSubscription constructor to allow infinite subscriptionTimeoutS
public AlarmSubscription(
String alarmId, float subscriptionTimeoutS, float subscriptionFrequencyMs) {
this.alarmId = alarmId;
diff --git a/src/policy/src/main/java/eu/teraflow/policy/service/ServiceGateway.java b/src/policy/src/main/java/eu/teraflow/policy/service/ServiceGateway.java
index a7483f4f62ac798932ea64a3351536ffb4e5155d..b8764efe312bae8b395beb3a22b1d2a57346b37d 100644
--- a/src/policy/src/main/java/eu/teraflow/policy/service/ServiceGateway.java
+++ b/src/policy/src/main/java/eu/teraflow/policy/service/ServiceGateway.java
@@ -16,6 +16,7 @@
package eu.teraflow.policy.service;
+import eu.teraflow.policy.context.model.Empty;
import eu.teraflow.policy.context.model.Service;
import eu.teraflow.policy.context.model.ServiceId;
import io.smallrye.mutiny.Uni;
@@ -23,4 +24,6 @@ import io.smallrye.mutiny.Uni;
public interface ServiceGateway {
Uni<ServiceId> updateService(Service service);
+
+ Uni<Empty> recomputeConnections(Service service);
}
diff --git a/src/policy/src/main/java/eu/teraflow/policy/service/ServiceGatewayImpl.java b/src/policy/src/main/java/eu/teraflow/policy/service/ServiceGatewayImpl.java
index d59330fd2b756749c333e70547989bb443bd19d8..f8f38f7520f608b52f432e41b96e46448234a68e 100644
--- a/src/policy/src/main/java/eu/teraflow/policy/service/ServiceGatewayImpl.java
+++ b/src/policy/src/main/java/eu/teraflow/policy/service/ServiceGatewayImpl.java
@@ -17,16 +17,19 @@
package eu.teraflow.policy.service;
import eu.teraflow.policy.Serializer;
+import eu.teraflow.policy.context.model.Empty;
import eu.teraflow.policy.context.model.Service;
import eu.teraflow.policy.context.model.ServiceId;
import io.quarkus.grpc.GrpcClient;
import io.smallrye.mutiny.Uni;
import javax.enterprise.context.ApplicationScoped;
import javax.inject.Inject;
+import org.jboss.logging.Logger;
import service.MutinyServiceServiceGrpc.MutinyServiceServiceStub;
@ApplicationScoped
public class ServiceGatewayImpl implements ServiceGateway {
+ private static final Logger LOGGER = Logger.getLogger(ServiceGatewayImpl.class);
@GrpcClient("service")
MutinyServiceServiceStub streamingDelegateService;
@@ -47,4 +50,23 @@ public class ServiceGatewayImpl implements ServiceGateway {
.onItem()
.transform(serializer::deserialize);
}
+
+ @Override
+ public Uni<Empty> recomputeConnections(Service service) {
+ service.getServiceEndPointIds().clear();
+ service.getServiceConstraints().clear();
+ final var serializedService = serializer.serialize(service);
+ LOGGER.info(serializedService);
+
+ return streamingDelegateService
+ .recomputeConnections(serializedService)
+ .onItem()
+ .transform(serializer::deserializeEmpty)
+ .onFailure()
+ .invoke(
+ x -> {
+ LOGGER.info("recomputeConnections failed with:");
+ LOGGER.info(x);
+ });
+ }
}
diff --git a/src/policy/src/main/java/eu/teraflow/policy/service/ServiceService.java b/src/policy/src/main/java/eu/teraflow/policy/service/ServiceService.java
index bbf91aff0c7ffeeb3f262fa3cebd582acda74502..aa4411ee9d79ea56c3d6f5b901f92b0125fe4b00 100644
--- a/src/policy/src/main/java/eu/teraflow/policy/service/ServiceService.java
+++ b/src/policy/src/main/java/eu/teraflow/policy/service/ServiceService.java
@@ -16,6 +16,7 @@
package eu.teraflow.policy.service;
+import eu.teraflow.policy.context.model.Empty;
import eu.teraflow.policy.context.model.Service;
import eu.teraflow.policy.context.model.ServiceId;
import io.smallrye.mutiny.Uni;
@@ -23,4 +24,6 @@ import io.smallrye.mutiny.Uni;
public interface ServiceService {
Uni<ServiceId> updateService(Service service);
+
+ Uni<Empty> recomputeConnections(Service service);
}
diff --git a/src/policy/src/main/java/eu/teraflow/policy/service/ServiceServiceImpl.java b/src/policy/src/main/java/eu/teraflow/policy/service/ServiceServiceImpl.java
index 74728924f70070e7651ac1af4ad75aba8b325c0e..109bcba261d4464313d5b6f21ee2bdde76b17879 100644
--- a/src/policy/src/main/java/eu/teraflow/policy/service/ServiceServiceImpl.java
+++ b/src/policy/src/main/java/eu/teraflow/policy/service/ServiceServiceImpl.java
@@ -16,6 +16,7 @@
package eu.teraflow.policy.service;
+import eu.teraflow.policy.context.model.Empty;
import eu.teraflow.policy.context.model.Service;
import eu.teraflow.policy.context.model.ServiceId;
import io.smallrye.mutiny.Uni;
@@ -36,4 +37,9 @@ public class ServiceServiceImpl implements ServiceService {
public Uni<ServiceId> updateService(Service service) {
return serviceGateway.updateService(service);
}
+
+ @Override
+ public Uni<Empty> recomputeConnections(Service service) {
+ return serviceGateway.recomputeConnections(service);
+ }
}
diff --git a/src/policy/src/test/java/eu/teraflow/policy/SerializerTest.java b/src/policy/src/test/java/eu/teraflow/policy/SerializerTest.java
index f06c30204b874cd6be30cd1a906c5087412e9640..31823e3f6a4b101a8e4deab967723b61a4dd0e97 100644
--- a/src/policy/src/test/java/eu/teraflow/policy/SerializerTest.java
+++ b/src/policy/src/test/java/eu/teraflow/policy/SerializerTest.java
@@ -1914,8 +1914,7 @@ class SerializerTest {
ServiceStatusEnum.SLA_VIOLATED,
ContextOuterClass.ServiceStatusEnum.SERVICESTATUS_SLA_VIOLATED),
Arguments.of(
- ServiceStatusEnum.UPDATING,
- ContextOuterClass.ServiceStatusEnum.SERVICESTATUS_UPDATING),
+ ServiceStatusEnum.UPDATING, ContextOuterClass.ServiceStatusEnum.SERVICESTATUS_UPDATING),
Arguments.of(
ServiceStatusEnum.UNDEFINED,
ContextOuterClass.ServiceStatusEnum.SERVICESTATUS_UNDEFINED));