Commit 25f71f10 authored by Waleed Akbar's avatar Waleed Akbar
Browse files

Changes on KPI Value Writer and Telemetry Backend 

- Renamed the method to "KafkaKpiConsumer" to avoid conflict with the "KafkaConsumer" import in KpiApiWriter.
- Removed unnecessary imports in KpiWriterToProm.
- Added `get_service_port_grpc` call and imports in the Telemetry backend service.
- Added new libraries to `requirements.in` for Telemetry.
parent d1c99c6b

src/kpi_value_writer/service/KpiValueWriter.py

+3 −3
Original line number Original line Diff line number Diff line
@@ -41,18 +41,18 @@ class KpiValueWriter(GenericGrpcService): 




    @staticmethod

    @staticmethod

    def RunKafkaConsumer():

    def RunKafkaConsumer():

        thread = threading.Thread(target=KpiValueWriter.KafkaConsumer, args=())

        thread = threading.Thread(target=KpiValueWriter.KafkaKpiConsumer, args=())

        ACTIVE_CONSUMERS.append(thread)

        ACTIVE_CONSUMERS.append(thread)

        thread.start()

        thread.start()





    @staticmethod

    @staticmethod

    def KafkaConsumer():

    def KafkaKpiConsumer():

        kpi_manager_client = KpiManagerClient()

        kpi_manager_client = KpiManagerClient()

        metric_writer      = MetricWriterToPrometheus()

        metric_writer      = MetricWriterToPrometheus()





        kafka_consumer  = KafkaConsumer(

        kafka_consumer  = KafkaConsumer(

            { 'bootstrap.servers' : KafkaConfig.get_kafka_address(),

            { 'bootstrap.servers' : KafkaConfig.get_kafka_address(),

              'group.id'          : __class__,

              'group.id'          : 'KpiValueWriter',

              'auto.offset.reset' : 'latest'}

              'auto.offset.reset' : 'latest'}

        )

        )

        kafka_consumer.subscribe([KafkaTopic.VALUE.value])

        kafka_consumer.subscribe([KafkaTopic.VALUE.value])

src/kpi_value_writer/service/MetricWriterToPrometheus.py

+4 −4
Original line number Original line Diff line number Diff line
@@ -14,11 +14,11 @@ 




# read Kafka stream from Kafka topic

# read Kafka stream from Kafka topic





import ast

# import ast

import time

# import time

import threading

# import threading

import logging

import logging

from prometheus_client import Gauge, CollectorRegistry

from prometheus_client import Gauge

from common.proto.kpi_sample_types_pb2 import KpiSampleType

from common.proto.kpi_sample_types_pb2 import KpiSampleType





from common.proto.kpi_value_api_pb2 import KpiValue

from common.proto.kpi_value_api_pb2 import KpiValue

src/telemetry/backend/requirements.in

+4 −0
Original line number Original line Diff line number Diff line
@@ -13,3 +13,7 @@ 
# limitations under the License.

# limitations under the License.





confluent-kafka==2.3.*

confluent-kafka==2.3.*

psycopg2-binary==2.9.*

SQLAlchemy==1.4.*

sqlalchemy-cockroachdb==1.4.*

SQLAlchemy-Utils==0.38.*
 
 No newline at end of file

src/telemetry/backend/service/TelemetryBackendService.py

+6 −2
Original line number Original line Diff line number Diff line
@@ -18,10 +18,12 @@ import random 
import logging

import logging

import threading

import threading

from typing import Any, Dict

from typing import Any, Dict

from common.proto.context_pb2 import Empty

# from common.proto.context_pb2 import Empty

from confluent_kafka import Producer as KafkaProducer

from confluent_kafka import Producer as KafkaProducer

from confluent_kafka import Consumer as KafkaConsumer

from confluent_kafka import Consumer as KafkaConsumer

from confluent_kafka import KafkaError

from confluent_kafka import KafkaError

from common.Constants import ServiceNameEnum

from common.Settings import get_service_port_grpc

from common.tools.kafka.Variables import KafkaConfig, KafkaTopic

from common.tools.kafka.Variables import KafkaConfig, KafkaTopic

from common.method_wrappers.Decorator import MetricsPool

from common.method_wrappers.Decorator import MetricsPool

from common.tools.service.GenericGrpcService import GenericGrpcService

from common.tools.service.GenericGrpcService import GenericGrpcService
@@ -38,8 +40,10 @@ class TelemetryBackendService(GenericGrpcService): 
    Class listens for request on Kafka topic, fetches requested metrics from device.

    Class listens for request on Kafka topic, fetches requested metrics from device.

    Produces metrics on both RESPONSE and VALUE kafka topics.

    Produces metrics on both RESPONSE and VALUE kafka topics.

    """

    """

    def __init__(self):

    def __init__(self, cls_name : str = __name__) -> None:

        LOGGER.info('Init TelemetryBackendService')

        LOGGER.info('Init TelemetryBackendService')

        port = get_service_port_grpc(ServiceNameEnum.TELEMETRYBACKEND)

        super().__init__(port, cls_name=cls_name)

        self.kafka_producer = KafkaProducer({'bootstrap.servers' : KafkaConfig.get_kafka_address()})

        self.kafka_producer = KafkaProducer({'bootstrap.servers' : KafkaConfig.get_kafka_address()})

        self.kafka_consumer = KafkaConsumer({'bootstrap.servers' : KafkaConfig.get_kafka_address(),

        self.kafka_consumer = KafkaConsumer({'bootstrap.servers' : KafkaConfig.get_kafka_address(),

                                            'group.id'           : 'backend',

                                            'group.id'           : 'backend',