Commit 4a11d913 authored by Fotis Soldatos's avatar Fotis Soldatos
Browse files

Merge branch 'policy/implement-monitoringService-interface' into 'develop' 

Policy/implement monitoring service interface

See merge request teraflow-h2020/controller!146
parents 57e16fed 15915e3b

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

+68 −0
Original line number Diff line number Diff line
@@ -56,6 +56,7 @@ import eu.teraflow.policy.context.model.Device; 
import eu.teraflow.policy.context.model.DeviceConfig;

import eu.teraflow.policy.context.model.DeviceDriverEnum;

import eu.teraflow.policy.context.model.DeviceOperationalStatus;

import eu.teraflow.policy.context.model.Empty;

import eu.teraflow.policy.context.model.EndPoint;

import eu.teraflow.policy.context.model.EndPointId;

import eu.teraflow.policy.context.model.Event;
@@ -95,11 +96,15 @@ import eu.teraflow.policy.monitoring.model.KpiValue; 
import eu.teraflow.policy.monitoring.model.KpiValueRange;

import eu.teraflow.policy.monitoring.model.StringKpiValue;

import eu.teraflow.policy.monitoring.model.SubsDescriptor;

import java.util.ArrayList;

import java.util.List;

import java.util.stream.Collectors;

import javax.inject.Singleton;

import kpi_sample_types.KpiSampleTypes;

import monitoring.Monitoring;

import monitoring.Monitoring.AlarmID;

import monitoring.Monitoring.KpiId;

import monitoring.Monitoring.SubscriptionID;

import policy.Policy;

import policy.Policy.PolicyRuleId;

import policy.PolicyAction;
@@ -1538,6 +1543,21 @@ public class Serializer { 
        return new KpiValueRange(minKpiValue, maxKpiValue);

    }



    public AlarmID serializeAlarmId(String alarmId) {

        final var builder = Monitoring.AlarmID.newBuilder();



        final var serializedAlarmIdUuid = serializeUuid(alarmId);

        builder.setAlarmId(serializedAlarmIdUuid);



        return builder.build();

    }



    public String deserialize(AlarmID serializedAlarmId) {

        final var serializedAlarmIdUuid = serializedAlarmId.getAlarmId();



        return deserialize(serializedAlarmIdUuid);

    }



    public Monitoring.AlarmDescriptor serialize(AlarmDescriptor alarmDescriptor) {

        final var builder = Monitoring.AlarmDescriptor.newBuilder();
@@ -1637,6 +1657,21 @@ public class Serializer { 
        return new SubsDescriptor(kpiId, samplingDurationS, samplingIntervalS, startDate, endDate);

    }



    public SubscriptionID serializeSubscriptionIdId(String subscriptionId) {

        final var builder = Monitoring.SubscriptionID.newBuilder();



        final var serializedSubscriptionIdUuid = serializeUuid(subscriptionId);

        builder.setSubsId(serializedSubscriptionIdUuid);



        return builder.build();

    }



    public String deserialize(SubscriptionID serializedSubscriptionId) {

        final var serializedSubscriptionIdUuid = serializedSubscriptionId.getSubsId();



        return deserialize(serializedSubscriptionIdUuid);

    }



    public PolicyCondition.PolicyRuleCondition serialize(PolicyRuleCondition policyRuleCondition) {

        final var builder = PolicyCondition.PolicyRuleCondition.newBuilder();
@@ -1908,6 +1943,28 @@ public class Serializer { 
        return new Kpi(kpiId, timestamp, kpiValue);

    }



    public List<Monitoring.Kpi> serialize(List<Kpi> kpis) {

        List<Monitoring.Kpi> serializedKpis = new ArrayList<>();



        for (Kpi kpi : kpis) {

            final var serializedKpi = serialize(kpi);



            serializedKpis.add(serializedKpi);

        }

        return serializedKpis;

    }



    public List<Kpi> deserialize(List<Monitoring.Kpi> serializedKpis) {

        List<Kpi> kpis = new ArrayList<>();



        for (Monitoring.Kpi serializedKpi : serializedKpis) {

            final var kpi = deserialize(serializedKpi);



            kpis.add(kpi);

        }

        return kpis;

    }



    public Monitoring.KpiDescriptor serialize(KpiDescriptor kpiDescriptor) {

        final var builder = Monitoring.KpiDescriptor.newBuilder();
@@ -2130,6 +2187,17 @@ public class Serializer { 
                deviceEndPoints);

    }



    public ContextOuterClass.Empty serializeEmpty(Empty empty) {



        final var builder = ContextOuterClass.Empty.newBuilder();



        return builder.build();

    }



    public Empty deserializeEmpty(ContextOuterClass.Empty serializedEmpty) {

        return new Empty();

    }



    public Uuid serializeUuid(String uuid) {

        return Uuid.newBuilder().setUuid(uuid).build();

    }

src/policy/src/main/java/eu/teraflow/policy/context/model/Empty.java

0 → 100644
+24 −0
Original line number Diff line number Diff line 
/*

* Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/)

*

* Licensed under the Apache License, Version 2.0 (the "License");

* you may not use this file except in compliance with the License.

* You may obtain a copy of the License at

*

*      http://www.apache.org/licenses/LICENSE-2.0

*

* Unless required by applicable law or agreed to in writing, software

* distributed under the License is distributed on an "AS IS" BASIS,

* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

* See the License for the specific language governing permissions and

* limitations under the License.

*/



package eu.teraflow.policy.context.model;



public class Empty {



    public Empty() {

        // Empty constructor to represent the Empty rpc message of context service

    }

}

src/policy/src/main/java/eu/teraflow/policy/monitoring/MonitoringGateway.java

0 → 100644
+48 −0
Original line number Diff line number Diff line 
/*

* Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/)

*

* Licensed under the Apache License, Version 2.0 (the "License");

* you may not use this file except in compliance with the License.

* You may obtain a copy of the License at

*

*      http://www.apache.org/licenses/LICENSE-2.0

*

* Unless required by applicable law or agreed to in writing, software

* distributed under the License is distributed on an "AS IS" BASIS,

* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

* See the License for the specific language governing permissions and

* limitations under the License.

*/



package eu.teraflow.policy.monitoring;



import eu.teraflow.policy.context.model.Empty;

import eu.teraflow.policy.monitoring.model.AlarmDescriptor;

import eu.teraflow.policy.monitoring.model.AlarmResponse;

import eu.teraflow.policy.monitoring.model.Kpi;

import eu.teraflow.policy.monitoring.model.KpiDescriptor;

import eu.teraflow.policy.monitoring.model.SubsDescriptor;

import io.smallrye.mutiny.Multi;

import io.smallrye.mutiny.Uni;

import java.util.List;



public interface MonitoringGateway {



    Uni<String> createKpi(KpiDescriptor kpiDescriptor);



    Uni<KpiDescriptor> getKpiDescriptor(String kpiId);



    Multi<List<Kpi>> subscribeKpi(SubsDescriptor subsDescriptor);



    Uni<SubsDescriptor> getSubsDescriptor(String subscriptionId);



    Uni<Empty> editKpiSubscription(SubsDescriptor subsDescriptor);



    Uni<String> createKpiAlarm(AlarmDescriptor alarmDescriptor);



    Uni<Empty> editKpiAlarm(AlarmDescriptor alarmDescriptor);



    Uni<AlarmDescriptor> getAlarmDescriptor(String alarmId);



    Multi<AlarmResponse> getAlarmResponseStream(String alarmId);

}

src/policy/src/main/java/eu/teraflow/policy/monitoring/MonitoringGatewayImpl.java

0 → 100644
+136 −0
Original line number Diff line number Diff line 
/*

* Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/)

*

* Licensed under the Apache License, Version 2.0 (the "License");

* you may not use this file except in compliance with the License.

* You may obtain a copy of the License at

*

*      http://www.apache.org/licenses/LICENSE-2.0

*

* Unless required by applicable law or agreed to in writing, software

* distributed under the License is distributed on an "AS IS" BASIS,

* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

* See the License for the specific language governing permissions and

* limitations under the License.

*/



package eu.teraflow.policy.monitoring;



import eu.teraflow.policy.Serializer;

import eu.teraflow.policy.context.model.Empty;

import eu.teraflow.policy.monitoring.model.AlarmDescriptor;

import eu.teraflow.policy.monitoring.model.AlarmResponse;

import eu.teraflow.policy.monitoring.model.Kpi;

import eu.teraflow.policy.monitoring.model.KpiDescriptor;

import eu.teraflow.policy.monitoring.model.SubsDescriptor;

import io.quarkus.grpc.GrpcClient;

import io.smallrye.mutiny.Multi;

import io.smallrye.mutiny.Uni;

import java.util.List;

import javax.enterprise.context.ApplicationScoped;

import javax.inject.Inject;

import monitoring.MutinyMonitoringServiceGrpc.MutinyMonitoringServiceStub;



@ApplicationScoped

public class MonitoringGatewayImpl implements MonitoringGateway {



    @GrpcClient("monitoring")

    MutinyMonitoringServiceStub streamingDelegateMonitoring;



    private final Serializer serializer;



    @Inject

    public MonitoringGatewayImpl(Serializer serializer) {

        this.serializer = serializer;

    }



    @Override

    public Uni<String> createKpi(KpiDescriptor kpiDescriptor) {

        final var serializedKpiDescriptor = serializer.serialize(kpiDescriptor);



        return streamingDelegateMonitoring

                .createKpi(serializedKpiDescriptor)

                .onItem()

                .transform(serializer::deserialize);

    }



    @Override

    public Uni<KpiDescriptor> getKpiDescriptor(String kpiId) {

        final var serializedKpiId = serializer.serializeKpiId(kpiId);



        return streamingDelegateMonitoring

                .getKpiDescriptor(serializedKpiId)

                .onItem()

                .transform(serializer::deserialize);

    }



    @Override

    public Multi<List<Kpi>> subscribeKpi(SubsDescriptor subsDescriptor) {

        final var serializedSubsDescriptor = serializer.serialize(subsDescriptor);



        return streamingDelegateMonitoring

                .subscribeKpi(serializedSubsDescriptor)

                .onItem()

                .transform(kpiList -> serializer.deserialize(kpiList.getKpiListList()));

    }



    @Override

    public Uni<SubsDescriptor> getSubsDescriptor(String subscriptionId) {

        final var serializedSubscriptionId = serializer.serializeSubscriptionIdId(subscriptionId);



        return streamingDelegateMonitoring

                .getSubsDescriptor(serializedSubscriptionId)

                .onItem()

                .transform(serializer::deserialize);

    }



    @Override

    public Uni<Empty> editKpiSubscription(SubsDescriptor subsDescriptor) {

        final var serializedSubsDescriptor = serializer.serialize(subsDescriptor);



        return streamingDelegateMonitoring

                .editKpiSubscription(serializedSubsDescriptor)

                .onItem()

                .transform(serializer::deserializeEmpty);

    }



    @Override

    public Uni<String> createKpiAlarm(AlarmDescriptor alarmDescriptor) {

        final var serializedAlarmDescriptor = serializer.serialize(alarmDescriptor);



        return streamingDelegateMonitoring

                .createKpiAlarm(serializedAlarmDescriptor)

                .onItem()

                .transform(serializer::deserialize);

    }



    @Override

    public Uni<Empty> editKpiAlarm(AlarmDescriptor alarmDescriptor) {

        final var serializedAlarmDescriptor = serializer.serialize(alarmDescriptor);



        return streamingDelegateMonitoring

                .editKpiAlarm(serializedAlarmDescriptor)

                .onItem()

                .transform(serializer::deserializeEmpty);

    }



    @Override

    public Uni<AlarmDescriptor> getAlarmDescriptor(String alarmId) {

        final var serializedAlarmId = serializer.serializeAlarmId(alarmId);



        return streamingDelegateMonitoring

                .getAlarmDescriptor(serializedAlarmId)

                .onItem()

                .transform(serializer::deserialize);

    }



    @Override

    public Multi<AlarmResponse> getAlarmResponseStream(String alarmId) {

        final var serializedAlarmId = serializer.serializeAlarmId(alarmId);



        return streamingDelegateMonitoring

                .getAlarmResponseStream(serializedAlarmId)

                .onItem()

                .transform(serializer::deserialize);

    }

}

src/policy/src/main/java/eu/teraflow/policy/monitoring/MonitoringService.java

0 → 100644
+48 −0
Original line number Diff line number Diff line 
/*

* Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/)

*

* Licensed under the Apache License, Version 2.0 (the "License");

* you may not use this file except in compliance with the License.

* You may obtain a copy of the License at

*

*      http://www.apache.org/licenses/LICENSE-2.0

*

* Unless required by applicable law or agreed to in writing, software

* distributed under the License is distributed on an "AS IS" BASIS,

* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

* See the License for the specific language governing permissions and

* limitations under the License.

*/



package eu.teraflow.policy.monitoring;



import eu.teraflow.policy.context.model.Empty;

import eu.teraflow.policy.monitoring.model.AlarmDescriptor;

import eu.teraflow.policy.monitoring.model.AlarmResponse;

import eu.teraflow.policy.monitoring.model.Kpi;

import eu.teraflow.policy.monitoring.model.KpiDescriptor;

import eu.teraflow.policy.monitoring.model.SubsDescriptor;

import io.smallrye.mutiny.Multi;

import io.smallrye.mutiny.Uni;

import java.util.List;



public interface MonitoringService {



    Uni<String> createKpi(KpiDescriptor kpiDescriptor);



    Uni<KpiDescriptor> getKpiDescriptor(String kpiId);



    Multi<List<Kpi>> subscribeKpi(SubsDescriptor subsDescriptor);



    Uni<SubsDescriptor> getSubsDescriptor(String subscriptionId);



    Uni<Empty> editKpiSubscription(SubsDescriptor subsDescriptor);



    Uni<String> createKpiAlarm(AlarmDescriptor alarmDescriptor);



    Uni<Empty> editKpiAlarm(AlarmDescriptor alarmDescriptor);



    Uni<AlarmDescriptor> getAlarmDescriptor(String alarmId);



    Multi<AlarmResponse> getAlarmResponseStream(String alarmId);

}