"README.md" did not exist on "d80b6962efa3a938a9ea7b15cbd663ddfb105fea"
Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
#
# 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.
import logging, grpc
from typing import Tuple, Any
from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method
from common.tools.kafka.Variables import KafkaConfig, KafkaTopic
from common.proto.context_pb2 import Empty
from common.proto.kpi_value_api_pb2_grpc import KpiValueAPIServiceServicer
from common.proto.kpi_value_api_pb2 import KpiValueList, KpiValueFilter
from confluent_kafka import Producer as KafkaProducer
from .NameMapping import NameMapping
LOGGER = logging.getLogger(__name__)
METRICS_POOL = MetricsPool('KpiValueAPI', 'NBIgRPC')
class KpiValueApiServiceServicerImpl(KpiValueAPIServiceServicer):
def __init__(self, name_mapping : NameMapping):
LOGGER.debug('Init KpiValueApiService')
@safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
def StoreKpiValues(self, request: KpiValueList, grpc_context: grpc.ServicerContext
) -> Empty:
LOGGER.debug('StoreKpiValues: Received gRPC message object: {:}'.format(request))
producer_obj = KafkaProducer({'bootstrap.servers' : KafkaConfig.SERVER_IP.value})
for kpi_value in request.kpi_value_list:
kpi_value_to_produce : Tuple [str, Any, Any] = (
kpi_value.kpi_id.kpi_id, # kpi_value.kpi_id.kpi_id.uuid
kpi_value.timestamp, # kpi_value.timestamp.timestamp
kpi_value.kpi_value_type # kpi_value.kpi_value_type.(many options)
)
LOGGER.debug('KPI to produce is {:}'.format(kpi_value_to_produce))
msg_key = "gRPC-KpiValueApi" # str(__class__.__name__)
# write this KPI to Kafka
producer_obj.produce(KafkaTopic.VALUE.value,
key = msg_key,
value = str(kpi_value_to_produce),
callback = self.delivery_callback
)
producer_obj.flush()
return Empty()
@safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
def SelectKpiValues(self, request: KpiValueFilter, grpc_context: grpc.ServicerContext
) -> KpiValueList:
LOGGER.debug('SelectKpiValues: Received gRPC message object: {:}'.format(request))
def delivery_callback(self, err, msg):
if err: print(f'Message delivery failed: {err}')
else: print(f'Message delivered to topic {msg.topic()}')