# 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. from influx_line_protocol import Metric import socket import requests import json import sys import logging LOGGER = logging.getLogger(__name__) class MetricsDB(): def __init__(self, host, ilp_port, rest_port, table): self.socket=socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.host=host self.ilp_port=int(ilp_port) self.rest_port=rest_port self.table=table def write_KPI(self,time,kpi_id,kpi_sample_type,device_id,endpoint_id,service_id,kpi_value): self.socket.connect((self.host,self.ilp_port)) metric = Metric(self.table) metric.with_timestamp(time) metric.add_tag('kpi_id', kpi_id) metric.add_tag('kpi_sample_type', kpi_sample_type) metric.add_tag('device_id', device_id) metric.add_tag('endpoint_id', endpoint_id) metric.add_tag('service_id', service_id) metric.add_value('kpi_value', kpi_value) str_metric = str(metric) str_metric += "\n" self.socket.sendall((str_metric).encode()) self.socket.close() def run_query(self, sql_query): query_params = {'query': sql_query, 'fmt' : 'json'} url = f"http://{self.host}:{self.rest_port}/exec" try: response = requests.get(url, params=query_params) json_response = json.loads(response.text) LOGGER.info(json_response) except requests.exceptions.RequestException as e: LOGGER.info(f'Error: {e}', file=sys.stderr)