diff --git a/helm/.helmignore b/helm/.helmignore deleted file mode 100644 index 0e8a0eb..0000000 --- a/helm/.helmignore +++ /dev/null @@ -1,23 +0,0 @@ -# Patterns to ignore when building packages. -# This supports shell glob matching, relative path matching, and -# negation (prefixed with !). Only one pattern per line. -.DS_Store -# Common VCS dirs -.git/ -.gitignore -.bzr/ -.bzrignore -.hg/ -.hgignore -.svn/ -# Common backup files -*.swp -*.bak -*.tmp -*.orig -*~ -# Various IDEs -.project -.idea/ -*.tmproj -.vscode/ diff --git a/helm/Chart.yaml b/helm/Chart.yaml deleted file mode 100644 index 585b890..0000000 --- a/helm/Chart.yaml +++ /dev/null @@ -1,26 +0,0 @@ -apiVersion: v2 -name: sientia-module -description: A generic application helm for Kubernetes. Contains a set of templates - that can be used to deploy a generic sientia python module to a Kubernetes cluster, - based in some git repository and a generic docker image. - -# A chart can be either an 'application' or a 'library' chart. -# -# Application charts are a collection of templates that can be packaged into versioned archives -# to be deployed. -# -# Library charts provide useful utilities or functions for the chart developer. They're included as -# a dependency of application charts to inject those utilities and functions into the rendering -# pipeline. Library charts do not define any templates and therefore cannot be deployed. -type: application - -# This is the chart version. This version number should be incremented each time you make changes -# to the chart and its templates, including the app version. -# Versions are expected to follow Semantic Versioning (https://semver.org/) -version: 0.1.0 - -# This is the version number of the application being deployed. This version number should be -# incremented each time you make changes to the application. Versions are not expected to -# follow Semantic Versioning. They should reflect the version the application is using. -# It is recommended to use it with quotes. -appVersion: "1.16.0" diff --git a/helm/README.md b/helm/README.md deleted file mode 100644 index 795d9ac..0000000 --- a/helm/README.md +++ /dev/null @@ -1,73 +0,0 @@ -# How to create an image and deploy this helm chart: - -## Image creation: -### Generate your ssh key to Docker -``` -ssh-keygen -t ed25519 -C "docker-access" -f ~/.ssh/id_ed25519_docker -``` - -Add the public key to yout Git SSH keys in your git platform - -### Enable Docker BuildKit -``` -export DOCKER_BUILDKIT=1 -``` -or make it permanent: -``` -echo '{ "features": { "buildkit": true } }' | sudo tee /etc/docker/daemon.json -sudo systemctl restart docker -``` - -### Build and upload image -``` -make -C docker -``` - -## SSH secrets: - -### Create secret to your git ssh credentials -We need a dedicated keypair to access GitHub from a Kubernetes pod without exposing your user credentials or tokens. -``` -ssh-keygen -t ed25519 -f git_key -C "k8s-deploy-key" -N "" -``` -- ```-t ed25519```: modern, secure key type - -- ```-f git_key```: saves to git_key (private) and git_key.pub (public) - -- ```-C "..."```: comment to identify this key - -- ```-N ""```: no passphrase (for non-interactive use in pods) - -### Add the Public Key to GitHub (Read-Only Deploy Key) -Deploy keys allow read-only access to a specific repo. This avoids using user tokens or giving broad permissions. - -1. Copy your public key: -``` -cat git_key.pub -``` - -2. Go to your GitHub repo → Settings → Deploy Keys - -3. Click “Add deploy key” - -- Title: K8s ReadOnly - -- Key: paste your git_key.pub - -- ✅ Check “Allow read access” - -- ❌ Do NOT check “Allow write access” - -### Create a Kubernetes Secret from the Private Key -Kubernetes Secrets securely store sensitive data like private keys. This secret will later be mounted into your pod for Git to use. -``` -kubectl create secret generic git-ssh-key \ - --namespace sientia-opc \ - --from-file=ssh-privatekey=git_key \ - --type=kubernetes.io/ssh-auth -``` -- ```ssh-privatekey```: required key name for type kubernetes.io/ssh-auth - -- Secret is base64-encoded and stored in Kubernetes (not encrypted unless you're using an external secret manager or encryption at rest is enabled) - - diff --git a/helm/files/github_known_hosts b/helm/files/github_known_hosts deleted file mode 100644 index 10ef85a..0000000 --- a/helm/files/github_known_hosts +++ /dev/null @@ -1,3 +0,0 @@ -github.com ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABgQCj7ndNxQowgcQnjshcLrqPEiiphnt+VTTvDP6mHBL9j1aNUkY4Ue1gvwnGLVlOhGeYrnZaMgRK6+PKCUXaDbC7qtbW8gIkhL7aGCsOr/C56SJMy/BCZfxd1nWzAOxSDPgVsmerOBYfNqltV9/hWCqBywINIR+5dIg6JTJ72pcEpEjcYgXkE2YEFXV1JHnsKgbLWNlhScqb2UmyRkQyytRLtL+38TGxkxCflmO+5Z8CSSNY7GidjMIZ7Q4zMjA2n1nGrlTDkzwDCsw+wqFPGQA179cnfGWOWRVruj16z6XyvxvjJwbz0wQZ75XK5tKSb7FNyeIEs4TT4jk+S4dhPeAUC5y+bDYirYgM4GC7uEnztnZyaVWQ7B381AK4Qdrwt51ZqExKbQpTUNn+EjqoTwvqNj4kqx5QUCI0ThS/YkOxJCXmPUWZbhjpCg56i+2aB6CmK2JGhn57K5mj0MNdBXA4/WnwH6XoPWJzK5Nyu2zB3nAZp+S5hpQs+p1vN1/wsjk= -github.com ecdsa-sha2-nistp256 AAAAE2VjZHNhLXNoYTItbmlzdHAyNTYAAAAIbmlzdHAyNTYAAABBBEmKSENjQEezOmxkZMy7opKgwFB9nkt5YRrYMjNuG5N87uRgg6CLrbo5wAdT/y6v0mKV0U2w0WZ2YB/++Tpockg= -github.com ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIOMqqnkVzrm0SdG6UOoqKLsabgH5C9okWi0dh2l9GKJl diff --git a/helm/templates/_helpers.tpl b/helm/templates/_helpers.tpl deleted file mode 100644 index f0a073a..0000000 --- a/helm/templates/_helpers.tpl +++ /dev/null @@ -1,62 +0,0 @@ -{{/* -Expand the name of the chart. -*/}} -{{- define "sientia-module.name" -}} -{{- default .Chart.Name .Values.nameOverride | trunc 63 | trimSuffix "-" }} -{{- end }} - -{{/* -Create a default fully qualified app name. -We truncate at 63 chars because some Kubernetes name fields are limited to this (by the DNS naming spec). -If release name contains chart name it will be used as a full name. -*/}} -{{- define "sientia-module.fullname" -}} -{{- if .Values.fullnameOverride }} -{{- .Values.fullnameOverride | trunc 63 | trimSuffix "-" }} -{{- else }} -{{- $name := default .Chart.Name .Values.nameOverride }} -{{- if contains $name .Release.Name }} -{{- .Release.Name | trunc 63 | trimSuffix "-" }} -{{- else }} -{{- printf "%s-%s" .Release.Name $name | trunc 63 | trimSuffix "-" }} -{{- end }} -{{- end }} -{{- end }} - -{{/* -Create chart name and version as used by the chart label. -*/}} -{{- define "sientia-module.chart" -}} -{{- printf "%s-%s" .Chart.Name .Chart.Version | replace "+" "_" | trunc 63 | trimSuffix "-" }} -{{- end }} - -{{/* -Common labels -*/}} -{{- define "sientia-module.labels" -}} -helm.sh/chart: {{ include "sientia-module.chart" . }} -{{ include "sientia-module.selectorLabels" . }} -{{- if .Chart.AppVersion }} -app.kubernetes.io/version: {{ .Chart.AppVersion | quote }} -{{- end }} -app.kubernetes.io/managed-by: {{ .Release.Service }} -{{- end }} - -{{/* -Selector labels -*/}} -{{- define "sientia-module.selectorLabels" -}} -app.kubernetes.io/name: {{ include "sientia-module.name" . }} -app.kubernetes.io/instance: {{ .Release.Name }} -{{- end }} - -{{/* -Create the name of the service account to use -*/}} -{{- define "sientia-module.serviceAccountName" -}} -{{- if .Values.serviceAccount.create }} -{{- default (include "sientia-module.fullname" .) .Values.serviceAccount.name }} -{{- else }} -{{- default "default" .Values.serviceAccount.name }} -{{- end }} -{{- end }} diff --git a/helm/templates/deployment.yaml b/helm/templates/deployment.yaml deleted file mode 100644 index a35f2b6..0000000 --- a/helm/templates/deployment.yaml +++ /dev/null @@ -1,96 +0,0 @@ -apiVersion: apps/v1 -kind: Deployment -metadata: - name: {{ .Values.nameOverride | default (include "sientia-module.fullname" .) }} - labels: - {{- include "sientia-module.labels" . | nindent 4 }} -spec: - {{- if not .Values.autoscaling.enabled }} - replicas: {{ .Values.replicaCount }} - {{- end }} - selector: - matchLabels: - {{- include "sientia-module.selectorLabels" . | nindent 6 }} - template: - metadata: - {{- with .Values.podAnnotations }} - annotations: - {{- toYaml . | nindent 8 }} - {{- end }} - labels: - {{- include "sientia-module.labels" . | nindent 8 }} - {{- with .Values.podLabels }} - {{- toYaml . | nindent 8 }} - {{- end }} - spec: - {{- with .Values.imagePullSecrets }} - imagePullSecrets: - {{- toYaml . | nindent 8 }} - {{- end }} - serviceAccountName: {{ include "sientia-module.serviceAccountName" . }} - {{- with .Values.podSecurityContext }} - securityContext: - {{- toYaml . | nindent 8 }} - {{- end }} - containers: - - name: {{ .Chart.Name }} - {{- with .Values.securityContext }} - securityContext: - {{- toYaml . | nindent 12 }} - {{- end }} - image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}" - imagePullPolicy: {{ .Values.image.pullPolicy }} - ports: - - name: http - containerPort: {{ .Values.service.port }} - protocol: TCP - env: - {{- toYaml .Values.env | nindent 12 }} - {{- with .Values.livenessProbe }} - livenessProbe: - {{- toYaml . | nindent 12 }} - {{- end }} - {{- with .Values.readinessProbe }} - readinessProbe: - {{- toYaml . | nindent 12 }} - {{- end }} - {{- with .Values.resources }} - resources: - {{- toYaml . | nindent 12 }} - {{- end }} - volumeMounts: - {{- if .Values.ssh.enabled }} - - name: ssh-key - mountPath: {{ .Values.ssh.sshPath }} - readOnly: true - - name: ssh-known-hosts - mountPath: {{ .Values.ssh.knownHostsPath }} - readOnly: true - {{- end }} - volumes: - {{- if .Values.ssh.enabled }} - - name: ssh-key - secret: - secretName: {{ .Values.ssh.secretName }} - items: - - key: ssh-privatekey - path: id_ed25519 - - name: ssh-known-hosts - configMap: - name: {{ include "sientia-module.fullname" . }}-ssh-known-hosts - items: - - key: known_hosts - path: known_hosts - {{- end }} - {{- with .Values.nodeSelector }} - nodeSelector: - {{- toYaml . | nindent 8 }} - {{- end }} - {{- with .Values.affinity }} - affinity: - {{- toYaml . | nindent 8 }} - {{- end }} - {{- with .Values.tolerations }} - tolerations: - {{- toYaml . | nindent 8 }} - {{- end }} diff --git a/helm/templates/hpa.yaml b/helm/templates/hpa.yaml deleted file mode 100644 index 2cbf74e..0000000 --- a/helm/templates/hpa.yaml +++ /dev/null @@ -1,32 +0,0 @@ -{{- if .Values.autoscaling.enabled }} -apiVersion: autoscaling/v2 -kind: HorizontalPodAutoscaler -metadata: - name: {{ include "sientia-module.fullname" . }} - labels: - {{- include "sientia-module.labels" . | nindent 4 }} -spec: - scaleTargetRef: - apiVersion: apps/v1 - kind: Deployment - name: {{ include "sientia-module.fullname" . }} - minReplicas: {{ .Values.autoscaling.minReplicas }} - maxReplicas: {{ .Values.autoscaling.maxReplicas }} - metrics: - {{- if .Values.autoscaling.targetCPUUtilizationPercentage }} - - type: Resource - resource: - name: cpu - target: - type: Utilization - averageUtilization: {{ .Values.autoscaling.targetCPUUtilizationPercentage }} - {{- end }} - {{- if .Values.autoscaling.targetMemoryUtilizationPercentage }} - - type: Resource - resource: - name: memory - target: - type: Utilization - averageUtilization: {{ .Values.autoscaling.targetMemoryUtilizationPercentage }} - {{- end }} -{{- end }} diff --git a/helm/templates/know-hosts-configmap.yaml b/helm/templates/know-hosts-configmap.yaml deleted file mode 100644 index ce38be3..0000000 --- a/helm/templates/know-hosts-configmap.yaml +++ /dev/null @@ -1,7 +0,0 @@ -apiVersion: v1 -kind: ConfigMap -metadata: - name: {{ include "sientia-module.fullname" . }}-ssh-known-hosts -data: - known_hosts: | -{{ .Files.Get "files/github_known_hosts" | indent 4 }} diff --git a/helm/templates/service.yaml b/helm/templates/service.yaml deleted file mode 100644 index 7859559..0000000 --- a/helm/templates/service.yaml +++ /dev/null @@ -1,15 +0,0 @@ -apiVersion: v1 -kind: Service -metadata: - name: {{ include "sientia-module.fullname" . }} - labels: - {{- include "sientia-module.labels" . | nindent 4 }} -spec: - type: {{ .Values.service.type }} - ports: - - port: {{ .Values.service.port }} - targetPort: http - protocol: TCP - name: http - selector: - {{- include "sientia-module.selectorLabels" . | nindent 4 }} diff --git a/helm/templates/serviceaccount.yaml b/helm/templates/serviceaccount.yaml deleted file mode 100644 index de4307b..0000000 --- a/helm/templates/serviceaccount.yaml +++ /dev/null @@ -1,13 +0,0 @@ -{{- if .Values.serviceAccount.create -}} -apiVersion: v1 -kind: ServiceAccount -metadata: - name: {{ include "sientia-module.serviceAccountName" . }} - labels: - {{- include "sientia-module.labels" . | nindent 4 }} - {{- with .Values.serviceAccount.annotations }} - annotations: - {{- toYaml . | nindent 4 }} - {{- end }} -automountServiceAccountToken: {{ .Values.serviceAccount.automount }} -{{- end }} diff --git a/ingestor/ingestor.py b/ingestor/ingestor.py index cb9e97e..3d39b35 100644 --- a/ingestor/ingestor.py +++ b/ingestor/ingestor.py @@ -2,6 +2,7 @@ from logging import Formatter, StreamHandler, getLogger from os import getenv from ingestor.managers.ingestor_manager import IngestorManager +from sientia_do.notifications.handlers import NotificationHandler class Ingestor: @@ -37,14 +38,27 @@ class Ingestor: self.poll_interval = int(getenv("POLL_INTERVAL", 5)) self.kafka_servers = kafka_servers.split(",") + self.logger = None self.init_logger() + self.notification_handler = NotificationHandler( + servers=self.kafka_servers, + logger=self.logger, + project_name="OPC_INGESTOR", + pipeline_name="-", + trigger_name="-", + model_name="-", + model="-" + ) + # build args for build notificarions components + + # call build notifications components def init_logger(self): """ Initializes a logger instance for the class. - This method sets up a logger with a specified log level, a stream handler, - and a formatter. The log level is determined by the environment variable - "LOG_LEVEL", defaulting to "INFO" if not set. The logger is then attached + This method sets up a logger with a specified log level, a stream handler, + and a formatter. The log level is determined by the environment variable + "LOG_LEVEL", defaulting to "INFO" if not set. The logger is then attached to the instance for use throughout the class. Attributes: self.logger (logging.Logger): The configured logger instance. @@ -105,7 +119,8 @@ class Ingestor: self.ingestor_manager = IngestorManager( self.kafka_servers, self.redis_host, self.redis_port, self.lease_ttl, - self.heartbeat_ttl, self.pod_id, self.poll_interval, self.logger, self.redis_username, self.redis_password + self.heartbeat_ttl, self.pod_id, self.poll_interval, self.logger, + self.notification_handler, self.redis_username, self.redis_password ) # Declare ingestor ative @@ -120,9 +135,9 @@ class Ingestor: def manage_no_slots(self, number_of_slots: int): """ Manages the scenario where there are no slots assigned to the ingestor. - This method checks if the ingestor is active (i.e., has no managed tags) - and if the number of available slots is greater than zero. If both - conditions are met, it attempts to acquire a slot lease and handles + This method checks if the ingestor is active (i.e., has no managed tags) + and if the number of available slots is greater than zero. If both + conditions are met, it attempts to acquire a slot lease and handles the acquired tags accordingly. Args: number_of_slots (int): The number of available slots. @@ -138,16 +153,16 @@ class Ingestor: def manage_leases(self, available_slots: int, lacking_ingestors: int, slot_diff: int): """ - Manages the allocation and deallocation of slot leases for ingestors based on + Manages the allocation and deallocation of slot leases for ingestors based on the number of available slots, lacking ingestors, and slot differences. Args: available_slots (int): The number of slots currently available for allocation. lacking_ingestors (int): The number of ingestors that are active and without slots. slot_diff (int): The difference between the total slots and the required slots. Behavior: - - If there are available slots and lacking ingestors, attempts to acquire slot leases + - If there are available slots and lacking ingestors, attempts to acquire slot leases for the available slots and processes the acquired tags. - - If there are no lacking ingestors but there are extra slots (slot_diff > 0), + - If there are no lacking ingestors but there are extra slots (slot_diff > 0), releases the extra slot leases to ensure proper allocation. Logs: - Logs the number of available slots when attempting to acquire leases. @@ -210,7 +225,7 @@ class Ingestor: self.manage_leases(available_slots, lacking_ingestors, slot_diff) - self.logger.info( + self.logger.debug( f"Active ingestors: {ingestors}, " f"Number of slots: {number_of_slots}, " f"Number of leases: {number_of_leases}, " @@ -223,3 +238,6 @@ class Ingestor: # Update opc servers self.ingestor_manager.update_slot_config() + + # Check OPC cycles + self.ingestor_manager.check_opc_servers_integrity() diff --git a/ingestor/managers/data_manager.py b/ingestor/managers/data_manager.py index 159b4a4..3f9ca3c 100644 --- a/ingestor/managers/data_manager.py +++ b/ingestor/managers/data_manager.py @@ -3,10 +3,14 @@ from logging import Logger from time import sleep from kafka import KafkaProducer from kafka.errors import NoBrokersAvailable +from sientia_do.notifications.handlers import NotificationHandler +from sientia_do.notifications.models import NotificationLevel +import traceback class DataManager(): - def __init__(self, kafka_servers: str, logger: Logger) -> None: + def __init__(self, kafka_servers: str, logger: Logger, + notification_handler: NotificationHandler) -> None: """ Initializes the DataManager instance with a Kafka producer. This constructor attempts to establish a connection to the specified Kafka servers @@ -45,6 +49,7 @@ class DataManager(): logger.info( f"DataManager initialized with Kafka servers: {kafka_servers}") self.logger = logger + self.notification_handler = notification_handler def __del__(self): """Destructor to close the producer connection.""" @@ -91,4 +96,12 @@ class DataManager(): self.kafka_producer.flush(timeout=10) except Exception as e: - self.logger.error(f"Failed to publish message: {e}") + trace = traceback.format_exc() + self.notification_handler.build_and_send_notification( + notification_id=f"KAFKA_PRODUCER_ERROR_{topic}", + message=f"Error publishing message to topic {topic}: {e}", + block="kafka_producer", + level=NotificationLevel.ERROR, + attachment_content=trace + ) + self.logger.error(trace) diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py index edd62ea..8c05c44 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/ingestor/managers/ingestor_manager.py @@ -5,16 +5,19 @@ from copy import deepcopy from ingestor.managers.data_manager import DataManager from ingestor.managers.opc_manager import OpcManager from ingestor.managers.resource_manager import ResourceManager +from sientia_do.notifications.models import Notification, NotificationLevel +from sientia_do.notifications.handlers import NotificationHandler class IngestorManager(): def __init__(self, kafka_servers: str, redis_host: str, redis_port: int, lease_ttl: int, heartbeat_ttl: int, pod_id: str, - poll_interval: int, logger: Logger, + poll_interval: int, logger: Logger, notification_handler: NotificationHandler, redis_username: str = None, redis_password: str = None): - self.data_manager = DataManager(kafka_servers, logger) + self.data_manager = DataManager( + kafka_servers, logger, notification_handler) self.opc_managers = {} self.resource_manager = ResourceManager( redis_host, redis_port, lease_ttl, heartbeat_ttl, pod_id, redis_username, redis_password @@ -25,6 +28,8 @@ class IngestorManager(): self.managed_tags = {} self.opc_servers = {} + self.notification_handler = notification_handler + def initialize_opc_from_config(self, server_config: dict, data_manager: DataManager, logger: Logger) -> OpcManager | None: """ Initializes an OPC Manager instance using the provided server configuration. @@ -49,15 +54,24 @@ class IngestorManager(): f"Initializing OpcManager at {server_config['url']}") manager = OpcManager( server_config['name'], server_config['url'], data_manager, logger, server_config['server_uri'], - server_config.get('cert_path'), server_config.get( + self.notification_handler, server_config.get('cert_path'), server_config.get( 'private_key_path'), server_config.get('server_cert_path') ) manager.config = server_config manager.connect() except Exception as e: - logger.error(f"Failed to initialize OpcManager: {e}") - traceback.print_exc() + + trace = traceback.format_exc() + self.notification_handler.build_and_send_notification( + notification_id=f'OPC_CONNECTION_ERROR_{server_config["name"]}', + message=f'Error initializing OPC manager: {e}', + block="opc_manager", + level=NotificationLevel.ERROR, + attachment_content=trace + ) + self.logger.error(trace) + return None return manager @@ -100,6 +114,7 @@ class IngestorManager(): ) elif self.opc_managers[server].config != server_config: + self.opc_managers[server].disconnect() del self.opc_managers[server] server_instance = self.initialize_opc_from_config( server_config, self.data_manager, self.logger @@ -117,6 +132,30 @@ class IngestorManager(): self.opc_managers[server].disconnect() self.opc_managers.pop(server, None) + def check_opc_servers_integrity(self): + """ + Checks the integrity of the OPC servers and updates the OPC servers if necessary. + """ + for server, opc_manager in self.opc_managers.items(): + opc_manager.check_cycles() + + is_lost = opc_manager.check_opc_listenning() + if is_lost: + self.logger.warning( + f"OPC server {server} is lost. " + f"Desconnecting from server." + ) + opc_manager.disconnect() + self.opc_managers[server] = self.initialize_opc_from_config( + opc_manager.config, self.data_manager, self.logger + ) + self.update_opc_servers() + for slot, slot_config in self.managed_tags.items(): + if server in slot_config: + self.manage_server( + slot, server, slot_config[server], slot_config[server]['tags'] + ) + def declare_active(self): """ Declares the ingestor as active by sending a heartbeat signal to the resource manager. @@ -342,10 +381,16 @@ class IngestorManager(): tags_to_sub ) except Exception as e: - self.logger.error( - f"Failed to subscribe to tags from {slot}:{server}\n{tags}: {e}" + trace = traceback.format_exc() + self.notification_handler.build_and_send_notification( + notification_id=f'OPC_SUBSCRIPTION_ERROR_{slot}:{server}', + message=f'Failed to subscribe to tags from {slot}:{server}\n{tags}: {e}', + block="opc_manager", + level=NotificationLevel.ERROR, + attachment_content=trace ) - self.logger.error(traceback.format_exc()) + self.logger.error(trace) + self.logger.warning( "Removing subscription from server " f"{server} for slot {slot}" diff --git a/ingestor/managers/opc_manager.py b/ingestor/managers/opc_manager.py index f376615..b8065f3 100644 --- a/ingestor/managers/opc_manager.py +++ b/ingestor/managers/opc_manager.py @@ -5,14 +5,14 @@ from typing import Callable from asyncua.crypto.security_policies import SecurityPolicyBasic256 from asyncua.sync import Client from sientia_do.notifications.models import NotificationLevel - +from sientia_do.notifications.handlers import NotificationHandler from ingestor.managers.data_manager import DataManager class OpcManager(): def __init__(self, name: str, url: str, data_manager: DataManager, - logger: Logger, server_uri: str, cert_path: str = None, - private_key_path: str = None, server_cert_path: str = None): + logger: Logger, server_uri: str, notification_handler: NotificationHandler, + cert_path: str = None, private_key_path: str = None, server_cert_path: str = None): self.url = url self.name = name self.server_uri = server_uri @@ -27,6 +27,8 @@ class OpcManager(): self.subscriptions = {} self.data_manager = data_manager + self.notification_handler = notification_handler + def __str__(self): return f"OpcManager(name={self.name}, url={self.url}, server_uri={self.server_uri})\n" \ f"nodes={self.nodes}, subscriptions={self.subscriptions}" @@ -220,6 +222,12 @@ class OpcManager(): source_timestamp = monitored_item.Value.SourceTimestamp tag = str(node) + self.logger.debug( + f"Data change notification received for tag: {tag} after {self.nodes[tag]['cycle_rule']['cycle_count']} cycles") + + self.logger.debug( + f"Resetting cycle count for tag: {tag} after {self.non_receive_count} OPC cycles") + self.nodes[tag]['cycle_rule']['cycle_count'] = 0 self.non_receive_count = 0 @@ -233,70 +241,48 @@ class OpcManager(): [self.data_manager.publish(e, data) for e in self.nodes[tag]['topics']] - def check_cycles(self, removed_data: dict, node: str, config: dict, - handle_listen_events: Callable[[str, str, NotificationLevel], None]): - """ - Checks the cycle count for a specific node and triggers a notification if the cycle count exceeds a threshold. - Args: - removed_data (dict): A dictionary containing data that has been removed. - Used to check if the node is present. - node (str): The identifier of the node being checked. - config (dict): Configuration dictionary containing metadata such as the tag name. - handle_listen_events (Callable[[str, str, NotificationLevel], None]): - A callback function to handle notification events. It takes three arguments: - - A message string describing the event. - - A tag string identifying the event. - - A NotificationLevel enum indicating the severity of the event. - Behavior: - - If the node is not in `removed_data`, the cycle count for the node is incremented. - - If the cycle count reaches or exceeds 5, the `handle_listen_events` callback is invoked - with a warning message, a tag, and a notification level. - Notification Example: - If the cycle count exceeds the threshold, a warning message is generated in the format: - "{cycles} cycles without receive from {node}:{name}" - where `cycles` is the current cycle count, `node` is the node identifier, and `name` is the tag name - from the `config` dictionary. + def check_cycles(self): """ + Checks the cycle counts for all monitored nodes and sends notifications if thresholds are exceeded. - if node not in removed_data.keys(): + This method iterates through all monitored nodes and updates their cycle counts based on + configured increments. If a node's cycle count exceeds a threshold (5 cycles), it triggers + a warning notification. + + """ + for node, config in self.nodes.items(): self.nodes[node]['cycle_rule']['cycle_count'] += self.nodes[node]['cycle_rule']['cycle_increment'] - if self.nodes[node]['cycle_rule']['cycle_count'] >= 5 and handle_listen_events: - name = config['tag_name'] - cycles = self.nodes[node]['cycle_rule']['cycle_count'] - handle_listen_events( - f'{cycles} cycles without receive from {node}:{name}', - f'TAG_{node}:{name}_LISTENNING_STOPPED', - NotificationLevel.WARNING - ) + if self.nodes[node]['cycle_rule']['cycle_count'] >= 5: + name = config['tag_name'] + cycles = self.nodes[node]['cycle_rule']['cycle_count'] + self.notification_handler.build_and_send_notification( + notification_id=f'TAG_{node}:{name}_LISTENNING_STOPPED', + message=f'{cycles} cycles without receive from {node}:{name}', + block="opc_manager", + level=NotificationLevel.WARNING + ) - def check_opc_listenning(self, handle_listen_events: Callable[[str, str, NotificationLevel], None]) -> None: + def check_opc_listenning(self) -> bool: """ - Monitors the OPC connection and triggers events based on the number of cycles - without receiving data from the OPC server. - Args: - handle_listen_events (Callable[[str, str, NotificationLevel], None]): - A callback function to handle notification events. It takes three arguments: - - A message string describing the event. - - An event code string. - - A NotificationLevel indicating the severity of the event. - Behavior: - - Increments the non-receive count each time the method is called. - - If the non-receive count reaches 5, triggers a notification event indicating - that the OPC server has stopped sending data. - - If the non-receive count reaches 15, triggers a notification event indicating - a retry to connect to the OPC server, disconnects the current session, and - reinitializes the collector with the existing configuration. + Checks the OPC connection and triggers notifications if the connection is lost. + Returns: + bool: True if the connection is lost, False otherwise. """ self.non_receive_count += 1 - if self.non_receive_count >= 5 and handle_listen_events: - handle_listen_events( - f'{self.non_receive_count} cycles without receive from OPC {self.name}. Tags: {json.dumps(self.nodes)}', - f'OPC_LISTENNING_STOPPED__{self.name}', NotificationLevel.ERROR) - if self.non_receive_count >= 15 and handle_listen_events: - handle_listen_events( - f'Retrying to connect to server {self.name}', - f'OPC_CONNECTION_RETRY__{self.name}', NotificationLevel.ERROR) - self.disconnect() - self.init_collector( - self.nodes, self.collect_period, self.period) + if self.non_receive_count >= 5: + self.notification_handler.build_and_send_notification( + notification_id=f'OPC_LISTENNING_STOPPED__{self.name}', + message=f'{self.non_receive_count} cycles without receive from OPC {self.name}. Tags: {json.dumps(self.nodes)}', + block="opc_manager", + level=NotificationLevel.ERROR + ) + if self.non_receive_count >= 15: + self.notification_handler.build_and_send_notification( + notification_id=f'OPC_CONNECTION_RETRY__{self.name}', + message=f'Retrying to connect to server {self.name}', + block="opc_manager", + level=NotificationLevel.ERROR + ) + return True + return False diff --git a/tests/unit/managers/test_data_manager.py b/tests/unit/managers/test_data_manager.py index 24df1b1..34c387a 100644 --- a/tests/unit/managers/test_data_manager.py +++ b/tests/unit/managers/test_data_manager.py @@ -1,7 +1,7 @@ from unittest.mock import ANY, MagicMock, patch from pytest import fixture from kafka.errors import NoBrokersAvailable - +from sientia_do.notifications.models import NotificationLevel from ingestor.managers.data_manager import DataManager @@ -10,7 +10,8 @@ from ingestor.managers.data_manager import DataManager def data_manager(kafka): return DataManager( kafka_servers="localhost:9092", - logger=MagicMock() + logger=MagicMock(), + notification_handler=MagicMock() ) @@ -20,7 +21,8 @@ def test___init___success(kafka): data_manager = DataManager( kafka_servers="localhost:9092", - logger=logger_mock + logger=logger_mock, + notification_handler=MagicMock() ) kafka.assert_called_once_with( @@ -46,7 +48,8 @@ def test___init___second_attempt(kafka): data_manager = DataManager( kafka_servers="localhost:9092", - logger=logger_mock + logger=logger_mock, + notification_handler=MagicMock() ) kafka.assert_any_call( @@ -79,7 +82,8 @@ def test___init___failure_max_attempts(kafka): try: DataManager( kafka_servers="localhost:9092", - logger=logger_mock + logger=logger_mock, + notification_handler=MagicMock() ) except NoBrokersAvailable as e: assert str( @@ -172,7 +176,8 @@ def test_publish(data_manager): data_manager.kafka_producer.flush.assert_called_once() -def test_publish_error(data_manager): +@patch("ingestor.managers.data_manager.traceback") +def test_publish_error(traceback, data_manager): topic = "test_topic" data = {"key": "value"} @@ -189,6 +194,10 @@ def test_publish_error(data_manager): ) # Check if the error was logged - data_manager.logger.error.assert_called_once_with( - "Failed to publish message: Test error" + data_manager.notification_handler.build_and_send_notification.assert_called_once_with( + notification_id=f"KAFKA_PRODUCER_ERROR_{topic}", + message=f"Error publishing message to topic {topic}: Test error", + block="kafka_producer", + level=NotificationLevel.ERROR, + attachment_content=traceback.format_exc.return_value ) diff --git a/tests/unit/managers/test_ingestor_manager.py b/tests/unit/managers/test_ingestor_manager.py index c326e81..6e59103 100644 --- a/tests/unit/managers/test_ingestor_manager.py +++ b/tests/unit/managers/test_ingestor_manager.py @@ -1,6 +1,6 @@ from unittest.mock import MagicMock, patch from pytest import fixture - +from sientia_do.notifications.models import NotificationLevel from ingestor.managers.ingestor_manager import IngestorManager @@ -16,14 +16,16 @@ def ingestor_manager(data_manager_mock, resource_manager_mock): heartbeat_ttl=60, pod_id="test_pod", poll_interval=5, - logger=MagicMock() + logger=MagicMock(), + notification_handler=MagicMock() ) @patch('ingestor.managers.ingestor_manager.OpcManager') @patch('ingestor.managers.ingestor_manager.DataManager') @patch('ingestor.managers.ingestor_manager.ResourceManager') -def test___init__(resource_manager_mock, data_manager_mock, opc_manager_mock): +@patch('ingestor.managers.ingestor_manager.NotificationHandler') +def test___init__(notification_handler_mock, resource_manager_mock, data_manager_mock, opc_manager_mock): ingestor = IngestorManager( kafka_servers="localhost:9092", @@ -33,15 +35,15 @@ def test___init__(resource_manager_mock, data_manager_mock, opc_manager_mock): heartbeat_ttl=60, pod_id="test_pod", poll_interval=5, - logger=MagicMock() + logger=MagicMock(), + notification_handler=MagicMock() ) opc_manager_mock.assert_not_called() data_manager_mock.assert_called_once_with( - "localhost:9092", ingestor.logger) -# resource_manager_mock.assert_called_once_with( -# "localhost", 6379, 60, 60, "test_pod") - + "localhost:9092", ingestor.logger, ingestor.notification_handler) + resource_manager_mock.assert_called_once_with( + "localhost", 6379, 60, 60, "test_pod", None, None) assert ingestor.poll_interval == 5 assert ingestor.managed_tags == {} assert ingestor.opc_servers == {} @@ -67,7 +69,8 @@ def test_initialize_opc_from_config(opc_manager, ingestor_manager): opc_manager.assert_called_once_with( server_config['name'], server_config['url'], ingestor_manager.data_manager, ingestor_manager.logger, - server_config['server_uri'], server_config['cert_path'], server_config['private_key_path'], + server_config['server_uri'], ingestor_manager.notification_handler, + server_config['cert_path'], server_config['private_key_path'], server_config['server_cert_path'] ) @@ -76,7 +79,8 @@ def test_initialize_opc_from_config(opc_manager, ingestor_manager): @patch('ingestor.managers.ingestor_manager.OpcManager') -def test_initialize_opc_from_config_exception(opc_manager, ingestor_manager): +@patch('ingestor.managers.ingestor_manager.traceback') +def test_initialize_opc_from_config_exception(traceback_mock, opc_manager, ingestor_manager): server_config = { 'name': 'server1', 'url': 'opc.tcp://localhost:4840', @@ -93,8 +97,15 @@ def test_initialize_opc_from_config_exception(opc_manager, ingestor_manager): server_config, ingestor_manager.data_manager, ingestor_manager.logger) assert result is None - ingestor_manager.logger.error.assert_called_once_with( - "Failed to initialize OpcManager: Initialization error") + + traceback_mock.format_exc.assert_called_once() + ingestor_manager.notification_handler.build_and_send_notification.assert_called_once_with( + notification_id=f'OPC_CONNECTION_ERROR_{server_config["name"]}', + message='Error initializing OPC manager: Initialization error', + block="opc_manager", + level=NotificationLevel.ERROR, + attachment_content=traceback_mock.format_exc.return_value + ) @patch('ingestor.managers.ingestor_manager.OpcManager') @@ -406,7 +417,8 @@ def test_manage_server(ingestor_manager): 'slot1', 'config1', ingestor_manager.poll_interval) -def test_manage_server_subscribe_failure(ingestor_manager): +@patch('ingestor.managers.ingestor_manager.traceback') +def test_manage_server_subscribe_failure(traceback_mock, ingestor_manager): ingestor_manager.opc_managers = { "server1": MagicMock(), "server2": MagicMock() @@ -431,9 +443,17 @@ def test_manage_server_subscribe_failure(ingestor_manager): 'slot1', 'config1', ingestor_manager.poll_interval) ingestor_manager.opc_managers["server1"].unsubscribe.assert_called_once_with( 'slot1') - ingestor_manager.logger.error.assert_any_call( - "Failed to subscribe to tags from slot1:server1\n{'tags': 'config1'}: Subscription error" + + traceback_mock.format_exc.assert_called_once() + + ingestor_manager.notification_handler.build_and_send_notification.assert_called_once_with( + notification_id='OPC_SUBSCRIPTION_ERROR_slot1:server1', + message='Failed to subscribe to tags from slot1:server1\n{\'tags\': \'config1\'}: Subscription error', + block="opc_manager", + level=NotificationLevel.ERROR, + attachment_content=traceback_mock.format_exc.return_value ) + ingestor_manager.logger.warning.assert_any_call( "Removing subscription from server server1 for slot slot1" ) @@ -475,3 +495,129 @@ def test_subscribe_to_tags(ingestor_manager): ingestor_manager.managed_tags['slot1'].pop.assert_called_once_with( 'server3', None) + + +def test_check_opc_servers_integrity_all_healthy(ingestor_manager): + # Setup mock OPC managers + opc_manager1 = MagicMock() + opc_manager1.check_cycles.return_value = None + opc_manager1.check_opc_listenning.return_value = False + opc_manager1.config = {"config": "config1"} + + opc_manager2 = MagicMock() + opc_manager2.check_cycles.return_value = None + opc_manager2.check_opc_listenning.return_value = False + opc_manager2.config = {"config": "config2"} + + ingestor_manager.opc_managers = { + "server1": opc_manager1, + "server2": opc_manager2 + } + + # Mock the initialize_opc_from_config method + ingestor_manager.initialize_opc_from_config = MagicMock() + + # Call the method + ingestor_manager.check_opc_servers_integrity() + + # Verify that check_cycles and check_opc_listenning were called for each server + opc_manager1.check_cycles.assert_called_once() + opc_manager1.check_opc_listenning.assert_called_once() + opc_manager2.check_cycles.assert_called_once() + opc_manager2.check_opc_listenning.assert_called_once() + + # Verify that no reinitialization was needed + ingestor_manager.initialize_opc_from_config.assert_not_called() + + +def test_check_opc_servers_integrity_server_lost(ingestor_manager): + # Setup mock OPC manager that will be lost + opc_manager = MagicMock() + opc_manager.check_cycles.return_value = None + opc_manager.check_opc_listenning.return_value = True # Server is lost + opc_manager.config = {"config": "config1"} + + ingestor_manager.opc_managers = { + "server1": opc_manager + } + + # Mock the initialize_opc_from_config method to return a new manager + new_manager = MagicMock() + ingestor_manager.initialize_opc_from_config = MagicMock( + return_value=new_manager) + + # Mock update_opc_servers and manage_server + ingestor_manager.update_opc_servers = MagicMock() + ingestor_manager.manage_server = MagicMock() + + # Call the method + ingestor_manager.check_opc_servers_integrity() + + # Verify that the lost server was disconnected + opc_manager.disconnect.assert_called_once() + + # Verify that a new manager was initialized + ingestor_manager.initialize_opc_from_config.assert_called_once_with( + opc_manager.config, ingestor_manager.data_manager, ingestor_manager.logger + ) + + # Verify that the new manager was assigned + assert ingestor_manager.opc_managers["server1"] == new_manager + + # Verify that update_opc_servers was called + ingestor_manager.update_opc_servers.assert_called_once() + + +def test_check_opc_servers_integrity_server_lost_with_tags(ingestor_manager): + # Setup mock OPC manager that will be lost + opc_manager = MagicMock() + opc_manager.check_cycles.return_value = None + opc_manager.check_opc_listenning.return_value = True # Server is lost + opc_manager.config = {"config": "config1"} + + ingestor_manager.opc_managers = { + "server1": opc_manager + } + + # Setup managed tags + ingestor_manager.managed_tags = { + "slot1": { + "server1": { + "config": "config1", + "tags": {"tag1": "value1"} + } + } + } + + # Mock the initialize_opc_from_config method to return a new manager + new_manager = MagicMock() + ingestor_manager.initialize_opc_from_config = MagicMock( + return_value=new_manager) + + # Mock update_opc_servers and manage_server + ingestor_manager.update_opc_servers = MagicMock() + ingestor_manager.manage_server = MagicMock() + + # Call the method + ingestor_manager.check_opc_servers_integrity() + + # Verify that the lost server was disconnected + opc_manager.disconnect.assert_called_once() + + # Verify that a new manager was initialized + ingestor_manager.initialize_opc_from_config.assert_called_once_with( + opc_manager.config, ingestor_manager.data_manager, ingestor_manager.logger + ) + + # Verify that the new manager was assigned + assert ingestor_manager.opc_managers["server1"] == new_manager + + # Verify that update_opc_servers was called + ingestor_manager.update_opc_servers.assert_called_once() + + # Verify that manage_server was called with the correct tags + ingestor_manager.manage_server.assert_called_once_with( + "slot1", "server1", + {"config": "config1", "tags": {"tag1": "value1"}}, + {"tag1": "value1"} + ) diff --git a/tests/unit/managers/test_opc_manager.py b/tests/unit/managers/test_opc_manager.py index 005d038..8886709 100644 --- a/tests/unit/managers/test_opc_manager.py +++ b/tests/unit/managers/test_opc_manager.py @@ -4,6 +4,7 @@ from unittest.mock import MagicMock, patch from pytest import fixture from asyncua.crypto.security_policies import SecurityPolicyBasic256 +import pytest from ingestor.managers.opc_manager import OpcManager from sientia_do.notifications.models import NotificationLevel @@ -36,7 +37,7 @@ tags = { def raw_opc_manager(): return OpcManager( 'TestConnector', 'opc.tcp://localhost:4840', MagicMock(), - MagicMock(), 'opc.tcp://localhost:4840' + MagicMock(), 'opc.tcp://localhost:4840', MagicMock() ) @@ -240,82 +241,102 @@ def test_datachange_notification(opc_manager_subscribed): assert opc_manager_subscribed.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] == 0 -def test_check_cycles(opc_manager_subscribed): - handler = MagicMock() - opc_manager_subscribed.nodes = tags - opc_manager_subscribed.nodes['ns=3;i=1001']['cycle_rule'] = { - 'cycle_increment': 1.0, - 'cycle_count': 0 +def test_check_cycles_no_notification(opc_manager): + # Setup: node with cycle_count just below threshold + opc_manager.nodes = { + 'ns=3;i=1001': { + 'tag_name': 'Counter', + 'cycle_rule': { + 'cycle_increment': 1.0, + 'cycle_count': 3.0 + } + } } - opc_manager_subscribed.check_cycles( - {}, 'ns=3;i=1001', tags['ns=3;i=1001'], handler) + opc_manager.notification_handler.build_and_send_notification = MagicMock() - assert opc_manager_subscribed.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] == 1 + opc_manager.check_cycles() - opc_manager_subscribed.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] = 0 - opc_manager_subscribed.check_cycles({'ns=3;i=1001': {}}, - 'ns=3;i=1001', tags['ns=3;i=1001'], handler) - - assert opc_manager_subscribed.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] == 0 - - opc_manager_subscribed.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] = 4 - opc_manager_subscribed.check_cycles( - {}, 'ns=3;i=1001', tags['ns=3;i=1001'], handler) - - assert opc_manager_subscribed.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] == 5 - handler.assert_called_once_with('5.0 cycles without receive from ns=3;i=1001:Counter', - 'TAG_ns=3;i=1001:Counter_LISTENNING_STOPPED', NotificationLevel.WARNING) + # After one increment, cycle_count = 4.0, still below threshold + assert opc_manager.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] == pytest.approx( + 4.0) + opc_manager.notification_handler.build_and_send_notification.assert_not_called() -def test_check_opc_listenning_5_cycles(opc_manager_subscribed): +def test_check_cycles_triggers_notification(opc_manager): + # Setup: node with cycle_count just below threshold, increment will cross threshold + opc_manager.nodes = { + 'ns=3;i=1001': { + 'tag_name': 'Counter', + 'cycle_rule': { + 'cycle_increment': 2.5, + 'cycle_count': 3.0 + } + } + } + opc_manager.notification_handler.build_and_send_notification = MagicMock() - handler = MagicMock() - opc_manager_subscribed.nodes = tags - opc_manager_subscribed.non_receive_count = 4 + opc_manager.check_cycles() - opc_manager_subscribed.check_opc_listenning(handler) - - handler.assert_called_once_with( - f'5 cycles without receive from OPC TestConnector. Tags: {json.dumps(tags)}', - 'OPC_LISTENNING_STOPPED__TestConnector', - NotificationLevel.ERROR + # After increment, cycle_count = 5.5, should trigger notification + assert opc_manager.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] == pytest.approx( + 5.5) + opc_manager.notification_handler.build_and_send_notification.assert_called_once_with( + notification_id='TAG_ns=3;i=1001:Counter_LISTENNING_STOPPED', + message='5.5 cycles without receive from ns=3;i=1001:Counter', + block="opc_manager", + level=NotificationLevel.WARNING ) -def test_check_opc_listenning_no_cycles(opc_manager_subscribed): - handler = MagicMock() - opc_manager_subscribed.non_receive_count = 0 +def test_check_opc_listenning_no_notification(opc_manager): + opc_manager.non_receive_count = 3 + opc_manager.notification_handler.build_and_send_notification = MagicMock() - opc_manager_subscribed.check_opc_listenning(handler) + result = opc_manager.check_opc_listenning() - handler.assert_not_called() + assert opc_manager.non_receive_count == 4 + opc_manager.notification_handler.build_and_send_notification.assert_not_called() + assert result is False -def test_check_opc_listenning_no_handler(opc_manager_subscribed): - opc_manager_subscribed.non_receive_count = 5 - opc_manager_subscribed.check_opc_listenning(None) +def test_check_opc_listenning_warning_notification(opc_manager): + opc_manager.non_receive_count = 4 + opc_manager.notification_handler.build_and_send_notification = MagicMock() - assert opc_manager_subscribed.non_receive_count == 6 + result = opc_manager.check_opc_listenning() - -def test_check_opc_listenning_15_cycles(opc_manager_subscribed): - handler = MagicMock() - opc_manager_subscribed.init_collector = MagicMock() - opc_manager_subscribed.non_receive_count = 14 - opc_manager_subscribed.collect_period = 1000 - opc_manager_subscribed.period = 500 - opc_manager_subscribed.nodes = tags - - opc_manager_subscribed.check_opc_listenning(handler) - handler.assert_any_call( - f'15 cycles without receive from OPC TestConnector. Tags: {json.dumps(tags)}', - 'OPC_LISTENNING_STOPPED__TestConnector', - NotificationLevel.ERROR + assert opc_manager.non_receive_count == 5 + opc_manager.notification_handler.build_and_send_notification.assert_called_once_with( + notification_id=f'OPC_LISTENNING_STOPPED__{opc_manager.name}', + message=f'5 cycles without receive from OPC {opc_manager.name}. Tags: {json.dumps(opc_manager.nodes)}', + block="opc_manager", + level=NotificationLevel.ERROR ) - handler.assert_any_call( - 'Retrying to connect to server TestConnector', - 'OPC_CONNECTION_RETRY__TestConnector', - NotificationLevel.ERROR + assert result is False + + +def test_check_opc_listenning_error_notification_and_retry(opc_manager): + opc_manager.non_receive_count = 14 + opc_manager.notification_handler.build_and_send_notification = MagicMock() + + result = opc_manager.check_opc_listenning() + + assert opc_manager.non_receive_count == 15 + # Should be called twice: once for 5, once for 15 + assert opc_manager.notification_handler.build_and_send_notification.call_count == 2 + calls = opc_manager.notification_handler.build_and_send_notification.call_args_list + # First call: 5 cycles warning + assert calls[0].kwargs == dict( + notification_id=f'OPC_LISTENNING_STOPPED__{opc_manager.name}', + message=f'15 cycles without receive from OPC {opc_manager.name}. Tags: {json.dumps(opc_manager.nodes)}', + block="opc_manager", + level=NotificationLevel.ERROR ) - opc_manager_subscribed.init_collector.assert_called_once_with( - tags, 1000, 500) + # Second call: 15 cycles retry + assert calls[1].kwargs == dict( + notification_id=f'OPC_CONNECTION_RETRY__{opc_manager.name}', + message=f'Retrying to connect to server {opc_manager.name}', + block="opc_manager", + level=NotificationLevel.ERROR + ) + assert result is True diff --git a/tests/unit/test_ingestor.py b/tests/unit/test_ingestor.py index 0639d81..448363c 100644 --- a/tests/unit/test_ingestor.py +++ b/tests/unit/test_ingestor.py @@ -1,46 +1,64 @@ -from unittest.mock import MagicMock, patch +from unittest.mock import ANY, MagicMock, patch from pytest import fixture from ingestor.ingestor import Ingestor -# @patch("ingestor.ingestor.getenv") -# @patch("ingestor.ingestor.Ingestor.init_logger") -# def test___init__(init_logger, getenv): -# getenv.side_effect = [ -# "localhost:9092,localhost:35", # KAFKA_SERVERS -# "localhost1", # REDIS_HOST -# '63790', # REDIS_PORT -# '100', # LEASE_TTL -# '200', # HEARTBEAT_TTL -# "localhost1", # HOSTNAME -# '50' # POLL_INTERVAL -# ] -# ingestor = Ingestor() +@patch("ingestor.ingestor.getenv") +@patch("ingestor.ingestor.Ingestor.init_logger") +@patch("ingestor.ingestor.NotificationHandler") +def test___init__(notification_handler, init_logger, getenv): + getenv.side_effect = [ + "localhost:9092,localhost:35", # KAFKA_SERVERS + "localhost1", # REDIS_HOST + '63790', # REDIS_PORT + "user", # REDIS_USERNAME + "password", # REDIS_PASSWORD + '100', # LEASE_TTL + '200', # HEARTBEAT_TTL + "localhost1", # HOSTNAME + '50' # POLL_INTERVAL + ] -# getenv.assert_any_call("KAFKA_SERVERS", "localhost:9092") -# getenv.assert_any_call("REDIS_HOST", "localhost") -# getenv.assert_any_call("REDIS_PORT", 6379) -# getenv.assert_any_call("LEASE_TTL", 10) -# getenv.assert_any_call("HEARTBEAT_TTL", 20) -# getenv.assert_any_call("HOSTNAME", "localhost") -# getenv.assert_any_call("POLL_INTERVAL", 5) + ingestor = Ingestor() -# assert ingestor.kafka_servers == ["localhost:9092", "localhost:35"] -# assert ingestor.redis_host == "localhost1" -# assert ingestor.redis_port == 63790 -# assert ingestor.lease_ttl == 100 -# assert ingestor.heartbeat_ttl == 200 -# assert ingestor.pod_id == "localhost1" -# assert ingestor.poll_interval == 50 + getenv.assert_any_call("KAFKA_SERVERS", "localhost:9092") + getenv.assert_any_call("REDIS_HOST", "localhost") + getenv.assert_any_call("REDIS_PORT", 6379) + getenv.assert_any_call("REDIS_USERNAME", None) + getenv.assert_any_call("REDIS_PASSWORD", None) + getenv.assert_any_call("LEASE_TTL", 10) + getenv.assert_any_call("HEARTBEAT_TTL", 20) + getenv.assert_any_call("HOSTNAME", "localhost") + getenv.assert_any_call("POLL_INTERVAL", 5) -# init_logger.assert_called_once() + assert ingestor.kafka_servers == ["localhost:9092", "localhost:35"] + assert ingestor.redis_host == "localhost1" + assert ingestor.redis_port == 63790 + assert ingestor.redis_username == "user" + assert ingestor.redis_password == "password" + assert ingestor.lease_ttl == 100 + assert ingestor.heartbeat_ttl == 200 + assert ingestor.pod_id == "localhost1" + assert ingestor.poll_interval == 50 + + init_logger.assert_called_once() + notification_handler.assert_called_once_with( + servers=["localhost:9092", "localhost:35"], + logger=ingestor.logger, + project_name="OPC_INGESTOR", + pipeline_name="-", + trigger_name="-", + model_name="-", + model="-" + ) @fixture @patch("ingestor.ingestor.getenv") @patch("ingestor.ingestor.Ingestor.init_logger") -def ingestor(init_logger, getenv): +@patch("ingestor.ingestor.NotificationHandler") +def ingestor(notification_handler, init_logger, getenv): ing = Ingestor() ing.logger = MagicMock() @@ -89,28 +107,31 @@ def test_handle_acquired_tags_success(ingestor_manager_started): ["tag1", "tag2"]) -# @patch("ingestor.ingestor.IngestorManager") -# def test_prepare_ingestor(ingestor_manager_mock, ingestor): -# ingestor_manager = ingestor_manager_mock.return_value -# ingestor_manager.get_slot_leases.return_value = True +@patch("ingestor.ingestor.IngestorManager") +def test_prepare_ingestor(ingestor_manager_mock, ingestor): + ingestor_manager = ingestor_manager_mock.return_value + ingestor_manager.get_slot_leases.return_value = True -# ingestor.prepare_ingestor() + ingestor.prepare_ingestor() -# ingestor_manager_mock.assert_called_once_with( -# ingestor.kafka_servers, -# ingestor.redis_host, -# ingestor.redis_port, -# ingestor.lease_ttl, -# ingestor.heartbeat_ttl, -# ingestor.pod_id, -# ingestor.poll_interval, -# ingestor.logger -# ) -# ingestor_manager.declare_active.assert_called_once() -# ingestor_manager.get_slot_leases.assert_called_once() + ingestor_manager_mock.assert_called_once_with( + ingestor.kafka_servers, + ingestor.redis_host, + ingestor.redis_port, + ingestor.lease_ttl, + ingestor.heartbeat_ttl, + ingestor.pod_id, + ingestor.poll_interval, + ingestor.logger, + ingestor.notification_handler, + ingestor.redis_username, + ingestor.redis_password + ) + ingestor_manager.declare_active.assert_called_once() + ingestor_manager.get_slot_leases.assert_called_once() -# ingestor.handle_acquired_tags( -# ingestor_manager.get_slot_leases.return_value) + ingestor.handle_acquired_tags( + ingestor_manager.get_slot_leases.return_value) def test_manage_slots_has_slots(ingestor_manager_started): @@ -194,53 +215,55 @@ def test_manage_leases_no_available_slots_extra_sltos(ingestor_manager_started): ["tag2", "tag3"]) -# def test_loop(ingestor_manager_started): -# ingestor_manager_started.manage_no_slots = MagicMock() -# ingestor_manager_started.manage_leases = MagicMock() -# ingestor_manager_started.ingestor_manager.managed_tags = { -# "slot1": "server1", -# "slot2": "server2", -# "slot3": "server3" -# } -# ingestor_manager_started.ingestor_manager.get_active_ingestors = MagicMock( -# return_value=["ingestor1", "ingestor2"]) -# ingestor_manager_started.ingestor_manager.get_number_of_slots = MagicMock( -# return_value=5) +def test_loop(ingestor_manager_started): + ingestor_manager_started.manage_no_slots = MagicMock() + ingestor_manager_started.manage_leases = MagicMock() + ingestor_manager_started.ingestor_manager.managed_tags = { + "slot1": "server1", + "slot2": "server2", + "slot3": "server3" + } + ingestor_manager_started.ingestor_manager.get_active_ingestors = MagicMock( + return_value=["ingestor1", "ingestor2"]) + ingestor_manager_started.ingestor_manager.get_number_of_slots = MagicMock( + return_value=5) + ingestor_manager_started.ingestor_manager.get_number_of_leases = MagicMock( + return_value=1) -# ingestor_manager_started.loop() + ingestor_manager_started.loop() -# ingestor_manager_started.ingestor_manager.declare_active.assert_called_once() -# ingestor_manager_started.ingestor_manager.get_active_ingestors.assert_called_once() -# ingestor_manager_started.ingestor_manager.get_number_of_slots.assert_called_once() + ingestor_manager_started.ingestor_manager.declare_active.assert_called_once() + ingestor_manager_started.ingestor_manager.get_active_ingestors.assert_called_once() + ingestor_manager_started.ingestor_manager.get_number_of_slots.assert_called_once() -# ingestor_manager_started.manage_no_slots.assert_called_once_with( -# ingestor_manager_started.ingestor_manager.get_number_of_slots.return_value) -# # Explanation: 5 - 2 = 3, 3 - 1 = 2 -# ingestor_manager_started.manage_leases.assert_called_once_with( -# 3, 2) -# ingestor_manager_started.ingestor_manager.update_slot_config.assert_called_once() + ingestor_manager_started.manage_no_slots.assert_called_once_with( + ingestor_manager_started.ingestor_manager.get_number_of_slots.return_value) + # Explanation: 5 - 2 = 3, 3 - 1 = 2 + ingestor_manager_started.manage_leases.assert_called_once_with( + 4, 3, 2) + ingestor_manager_started.ingestor_manager.update_slot_config.assert_called_once() -# def test_loop_no_managed(ingestor_manager_started): -# ingestor_manager_started.manage_no_slots = MagicMock() -# ingestor_manager_started.manage_leases = MagicMock() -# ingestor_manager_started.ingestor_manager.managed_tags = {} -# ingestor_manager_started.ingestor_manager.get_active_ingestors = MagicMock( -# return_value=["ingestor1", "ingestor2"]) -# ingestor_manager_started.ingestor_manager.get_number_of_slots = MagicMock( -# return_value=5) +def test_loop_no_managed(ingestor_manager_started): + ingestor_manager_started.manage_no_slots = MagicMock() + ingestor_manager_started.manage_leases = MagicMock() + ingestor_manager_started.ingestor_manager.managed_tags = {} + ingestor_manager_started.ingestor_manager.get_active_ingestors = MagicMock( + return_value=["ingestor1", "ingestor2"]) + ingestor_manager_started.ingestor_manager.get_number_of_slots = MagicMock( + return_value=5) -# ingestor_manager_started.loop() + ingestor_manager_started.loop() -# ingestor_manager_started.ingestor_manager.declare_active.assert_called_once() -# ingestor_manager_started.ingestor_manager.get_active_ingestors.assert_called_once() -# ingestor_manager_started.ingestor_manager.get_number_of_slots.assert_called_once() + ingestor_manager_started.ingestor_manager.declare_active.assert_called_once() + ingestor_manager_started.ingestor_manager.get_active_ingestors.assert_called_once() + ingestor_manager_started.ingestor_manager.get_number_of_slots.assert_called_once() -# ingestor_manager_started.manage_no_slots.assert_called_once_with( -# ingestor_manager_started.ingestor_manager.get_number_of_slots.return_value) -# # Explanation: 5 - 2 = 3, 3 - 1 = 2 -# ingestor_manager_started.manage_leases.assert_called_once_with( -# 3, -1) -# ingestor_manager_started.ingestor_manager.update_slot_config.assert_called_once() -# ingestor_manager_started.logger.info.assert_any_call( -# "No slots acquired in this loop") + ingestor_manager_started.manage_no_slots.assert_called_once_with( + ingestor_manager_started.ingestor_manager.get_number_of_slots.return_value) + # Explanation: 5 - 2 = 3, 3 - 1 = 2 + ingestor_manager_started.manage_leases.assert_called_once_with( + ANY, 3, -1) + ingestor_manager_started.ingestor_manager.update_slot_config.assert_called_once() + ingestor_manager_started.logger.info.assert_any_call( + "No slots acquired in this loop") diff --git a/helm/values.yaml b/values.yaml similarity index 96% rename from helm/values.yaml rename to values.yaml index 32cc058..08d9a14 100644 --- a/helm/values.yaml +++ b/values.yaml @@ -123,7 +123,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-opc-ingestor.git" - name: GITHUB_BRANCH - value: "SIENTIAPDE-1017-criar-helm-generico-para-os-modulos-do-sientia" + value: "SIENTIAPDE-988-criar-ingestor-opc" - name: PYTHON_APP value: "ingestor.app" @@ -145,11 +145,11 @@ env: name: redis key: redis-password - name: LEASE_TTL - value: "10" - - name: HEARTBEAT_TTL value: "20" + - name: HEARTBEAT_TTL + value: "30" - name: POLL_INTERVAL - value: "5" + value: "10" ssh: @@ -159,4 +159,4 @@ ssh: knownHostsPath: /mnt/known_hosts # kubectl create secret docker-registry docker-hub-secret --namespace sientia-opc --docker-server=http://aignosi.azurecr.io --docker-username=aignosi --docker-password=5I5zpQ6sRaHqX1hD3dr+2mo647yO3FRc359/wu6gsP+ACRDRz5mp -# helm upgrade --install sientia-dataops-opc-ingestor ./helm -n sientia-opc --create-namespace \ No newline at end of file +# helm upgrade --install sientia-dataops-opc-ingestor sientia/sientia-module -n sientia-opc --create-namespace -f ./values.yaml \ No newline at end of file