wip: policy component calls recomputeConnection rpc (9cfdce17) · Commits · TFS / controller

src/policy/src/main/java/eu/teraflow/policy/PolicyServiceImpl.java

+28 −21

Original line number	Original line	Diff line number	Diff line
	@@ -70,6 +70,8 @@ public class PolicyServiceImpl implements PolicyService {
	private static final int ACCEPTABLE_NUMBER_OF_ALARMS = 3;		private static final int ACCEPTABLE_NUMBER_OF_ALARMS = 3;
	private static final int MONITORING_WINDOW_IN_SECONDS = 5;		private static final int MONITORING_WINDOW_IN_SECONDS = 5;
	private static final int SAMPLING_RATE_PER_SECOND = 1;		private static final int SAMPLING_RATE_PER_SECOND = 1;
			// 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 =		private static final PolicyRuleState INSERTED_POLICYRULE_STATE =
	new PolicyRuleState(		new PolicyRuleState(
	@@ -213,6 +215,7 @@ public class PolicyServiceImpl implements PolicyService {
	LOGGER.infof("alarmDescriptor:");		LOGGER.infof("alarmDescriptor:");
	LOGGER.infof(alarmDescriptor.toString());		LOGGER.infof(alarmDescriptor.toString());
	alarmIds.add(monitoringService.setKpiAlarm(alarmDescriptor));		alarmIds.add(monitoringService.setKpiAlarm(alarmDescriptor));
			noAlarms = 0;
	}		}

	LOGGER.infof("Passed 7th");		LOGGER.infof("Passed 7th");
	@@ -229,8 +232,7 @@ public class PolicyServiceImpl implements PolicyService {
	id -> {		id -> {
	alarmPolicyRuleServiceMap.put(id, policyRuleService);		alarmPolicyRuleServiceMap.put(id, policyRuleService);

	var alarmSubscription = new AlarmSubscription(id, 60, 500);		var alarmSubscription = new AlarmSubscription(id, 60, 5000);
	LOGGER.infof("Creating Alarm Subscription with id: %s", id);
	return monitoringService.getAlarmResponseStream(alarmSubscription);		return monitoringService.getAlarmResponseStream(alarmSubscription);
	}));		}));
	}		}
	@@ -312,7 +314,7 @@ public class PolicyServiceImpl implements PolicyService {
	.subscribe()		.subscribe()
	.with(		.with(
	alarmId -> {		alarmId -> {
	alarmSubscriptionList.add(new AlarmSubscription(alarmId, 0, 0));		alarmSubscriptionList.add(new AlarmSubscription(alarmId, 30, 4000));
	});		});
	}		}

	@@ -737,13 +739,11 @@ public class PolicyServiceImpl implements PolicyService {
	serviceService.recomputeConnections(deserializedService)		serviceService.recomputeConnections(deserializedService)
	.subscribe()		.subscribe()
	.with( x -> {		.with( x -> {
	LOGGER.info("recomputeConnections failed with:");
	LOGGER.info(x);
	});
	LOGGER.info("called recomputeConnections with:");		LOGGER.info("called recomputeConnections with:");
	LOGGER.info(deserializedService);		LOGGER.info(deserializedService);
	setPolicyRuleServiceToContext(policyRuleService, ENFORCED_POLICYRULE_STATE);		setPolicyRuleServiceToContext(policyRuleService, ENFORCED_POLICYRULE_STATE);
	});		});
			});
	}		}

	private void applyActionService(String alarmId) {		private void applyActionService(String alarmId) {
	@@ -757,6 +757,8 @@ public class PolicyServiceImpl implements PolicyService {
	PolicyRuleAction test = policyRuleActionMap.get(alarmId);		PolicyRuleAction test = policyRuleActionMap.get(alarmId);
	LOGGER.info(test);		LOGGER.info(test);

			if (noAlarms == 0) {
			noAlarms++;
	setPolicyRuleServiceToContext(policyRuleService, ACTIVE_POLICYRULE_STATE);		setPolicyRuleServiceToContext(policyRuleService, ACTIVE_POLICYRULE_STATE);

	switch (policyRuleAction.getPolicyRuleActionEnum()) {		switch (policyRuleAction.getPolicyRuleActionEnum()) {
	@@ -770,6 +772,11 @@ public class PolicyServiceImpl implements PolicyService {
	LOGGER.errorf(INVALID_MESSAGE, policyRuleAction.getPolicyRuleActionEnum());		LOGGER.errorf(INVALID_MESSAGE, policyRuleAction.getPolicyRuleActionEnum());
	return;		return;
	}		}
			} else if (noAlarms == 5) {
			noAlarms = 0;
			} else {
			noAlarms++;
			}
	}		}

	private List<AlarmDescriptor> parsePolicyRuleCondition(PolicyRuleBasic policyRuleBasic) {		private List<AlarmDescriptor> parsePolicyRuleCondition(PolicyRuleBasic policyRuleBasic) {

src/policy/src/main/java/eu/teraflow/policy/service/ServiceGatewayImpl.java

+1 −0

Original line number	Original line	Diff line number	Diff line
	@@ -58,6 +58,7 @@ public class ServiceGatewayImpl implements ServiceGateway {
	//final var dummyService = new Service(tmpServiceId, null, null, null, null, null, 0);		//final var dummyService = new Service(tmpServiceId, null, null, null, null, null, 0);

	service.getServiceEndPointIds().clear();		service.getServiceEndPointIds().clear();
			service.getServiceConstraints().clear();
	final var serializedService = serializer.serialize(service);		final var serializedService = serializer.serialize(service);
	LOGGER.info(serializedService);		LOGGER.info(serializedService);

src/service/service/ServiceServiceServicerImpl.py

+2 −1

Original line number	Original line	Diff line number	Diff line
	@@ -174,6 +174,7 @@ class ServiceServiceServicerImpl(ServiceServiceServicer):

	@safe_and_metered_rpc_method(METRICS_POOL, LOGGER)		@safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
	def RecomputeConnections(self, request : Service, context : grpc.ServicerContext) -> Empty:		def RecomputeConnections(self, request : Service, context : grpc.ServicerContext) -> Empty:
			LOGGER.info("************************************Called RecomputeConnections*******************************************")
	if len(request.service_endpoint_ids) > 0:		if len(request.service_endpoint_ids) > 0:
	raise NotImplementedException('update-endpoints')		raise NotImplementedException('update-endpoints')

	@@ -254,7 +255,7 @@ class ServiceServiceServicerImpl(ServiceServiceServicer):
	pathcomp_request = PathCompRequest()		pathcomp_request = PathCompRequest()
	pathcomp_request.services.append(updated_service_with_uuids)		pathcomp_request.services.append(updated_service_with_uuids)
	#pathcomp_request.k_disjoint_path.num_disjoint = 100		#pathcomp_request.k_disjoint_path.num_disjoint = 100
	pathcomp_request.k_shortest_path.k_inspection = 100		pathcomp_request.k_shortest_path.k_inspection = 4
	pathcomp_request.k_shortest_path.k_return = 3		pathcomp_request.k_shortest_path.k_return = 3

	LOGGER.debug('pathcomp_request={:s}'.format(grpc_message_to_json_string(pathcomp_request)))		LOGGER.debug('pathcomp_request={:s}'.format(grpc_message_to_json_string(pathcomp_request)))