Merge pull request #352 from SUNET/bugfix.refresh_job_events

Bugfix.refresh job events

alembic/versions/2f9faee221a7_add_ipv6_gw_field_to_mgmtdomain.py

@@ -6,6 +6,7 @@
 
 """
 import sqlalchemy as sa
+
 from alembic import op
 
 # revision identifiers, used by Alembic.
@@ -18,5 +19,6 @@
 def upgrade():
     op.add_column("mgmtdomain", sa.Column("ipv6_gw", sa.Unicode(43)))
 
+
 def downgrade():
     op.drop_column("mgmtdomain", "ipv6_gw")

alembic/versions/395427a732d6_add_aborted_state_for_jobs.py

@@ -22,7 +22,6 @@ def upgrade():
     op.execute("ALTER TYPE jobstatus ADD VALUE 'ABORTED' AFTER 'EXCEPTION'")
 
 
-
 def downgrade():
     op.drop_index(op.f("ix_job_ticket_ref"), table_name="job")
     op.drop_index(op.f("ix_job_status"), table_name="job")

alembic/versions/5aa2593e147a_save_change_score_in_separate_column_.py

@@ -6,6 +6,7 @@
 
 """
 import sqlalchemy as sa
+
 from alembic import op
 
 # revision identifiers, used by Alembic.
@@ -18,5 +19,6 @@
 def upgrade():
     op.add_column("job", sa.Column("change_score", sa.SmallInteger(), nullable=True))
 
+
 def downgrade():
     op.drop_column("job", "change_score")

alembic/versions/6c6bec879fa8_add_joblock_database.py

@@ -6,6 +6,7 @@
 
 """
 import sqlalchemy as sa
+
 from alembic import op
 
 # revision identifiers, used by Alembic.

alembic/versions/8a635012afa7_add_new_interface_config_types.py

@@ -21,8 +21,6 @@ def upgrade():
     op.execute("ALTER TYPE interfaceconfigtype ADD VALUE 'ACCESS_DOWNLINK' AFTER 'ACCESS_UPLINK'")
 
 
-
 def downgrade():
     # removing extra types in an enum can make fields in the database invalid
     pass
-

alembic/versions/922589188efe_add_columns_for_dist_switch_vxlan_fabric.py

@@ -24,9 +24,8 @@ def upgrade():
     op.add_column(
         "device", sa.Column("oob_ip", sqlalchemy_utils.types.ip_address.IPAddressType(length=50), nullable=True)
     )
-    op.add_column(
-        "mgmtdomain", sa.Column("esi_mac", sa.String(length=12), nullable=True)
-    )
+    op.add_column("mgmtdomain", sa.Column("esi_mac", sa.String(length=12), nullable=True))
+
 
 def downgrade():
     op.drop_column("mgmtdomain", "esi_mac")

alembic/versions/9478bbaf8010_add_reservedip_table.py

@@ -31,13 +31,13 @@ def upgrade():
         sa.PrimaryKeyConstraint("device_id"),
     )
     op.create_index(op.f("ix_reservedip_device_id"), "reservedip", ["device_id"], unique=False)
-    #op.create_unique_constraint("jobid_unique1", "joblock", ["jobid"])
+    # op.create_unique_constraint("jobid_unique1", "joblock", ["jobid"])
     # ### end Alembic commands ###
 
 
 def downgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    #op.drop_constraint("jobid_unique1", "joblock", type_="unique")
+    # op.drop_constraint("jobid_unique1", "joblock", type_="unique")
     op.drop_index(op.f("ix_reservedip_device_id"), table_name="reservedip")
     op.drop_table("reservedip")
     # ### end Alembic commands ###

alembic/versions/9d01bce3c835_add_job_status_aborting_and_start_.py

@@ -19,13 +19,12 @@
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    
+
     op.add_column("job", sa.Column("start_arguments", postgresql.JSONB(astext_type=sa.Text()), nullable=True))
     op.execute("COMMIT")
     op.execute("ALTER TYPE jobstatus ADD VALUE 'ABORTING' AFTER 'ABORTED'")
 
 
-
 def downgrade():
     op.drop_column("job", "start_arguments")
     # removing extra types in an enum can make fields in the database invalid

alembic/versions/a3f3bc390462_alembic_start.py

@@ -21,4 +21,4 @@ def upgrade():
 def downgrade():
     # ### commands auto generated by Alembic - please adjust! ###
     pass
-    # ### end Alembic commands ###
+    # ### end Alembic commands ###

alembic/versions/d3aa4454ba7b_create_initial_database.py

@@ -5,13 +5,14 @@
 Create Date: 2024-01-22 13:00:27.673060
 
 """
-from alembic import op
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
 
+from alembic import op
+
 # revision identifiers, used by Alembic.
-revision = 'd3aa4454ba7b'
-down_revision = 'a3f3bc390462'
+revision = "d3aa4454ba7b"
+down_revision = "a3f3bc390462"
 branch_labels = None
 depends_on = None
 
@@ -37,10 +38,20 @@ def upgrade():
         sa.Column("vendor", sa.String(length=64)),
         sa.Column("model", sa.String(length=64)),
         sa.Column("os_version", sa.String(length=64)),
-        sa.Column("synchronized", sa.Boolean()),      
+        sa.Column("synchronized", sa.Boolean()),
         sa.Column(
             "state",
-            sa.Enum("UNKNOWN", "PRE_CONFIGURED", "DHCP_BOOT", "DISCOVERED", "INIT", "MANAGED", "MANAGED_NOIF", "UNMANAGED", name="devicestate"),
+            sa.Enum(
+                "UNKNOWN",
+                "PRE_CONFIGURED",
+                "DHCP_BOOT",
+                "DISCOVERED",
+                "INIT",
+                "MANAGED",
+                "MANAGED_NOIF",
+                "UNMANAGED",
+                name="devicestate",
+            ),
             nullable=False,
         ),
         sa.Column(
@@ -50,16 +61,26 @@ def upgrade():
         ),
         sa.Column("last_seen", sa.TIMESTAMP()),
         sa.UniqueConstraint("hostname"),
-        sa.ForeignKeyConstraint(["site_id"], ["site.id"])
-    )      
+        sa.ForeignKeyConstraint(["site_id"], ["site.id"]),
+    )
 
     op.create_table(
         "interface",
         sa.Column("device_id", sa.Integer(), nullable=False, primary_key=True),
         sa.Column("name", sa.String(length=255), nullable=False, primary_key=True),
         sa.Column(
             "configtype",
-            sa.Enum("UNKNOWN", "UNMANAGED", "CONFIGFILE", "CUSTOM", "ACCESS_AUTO", "ACCESS_UNTAGGED", "ACCESS_TAGGED", "ACCESS_UPLINK", name="interfaceconfigtype"),
+            sa.Enum(
+                "UNKNOWN",
+                "UNMANAGED",
+                "CONFIGFILE",
+                "CUSTOM",
+                "ACCESS_AUTO",
+                "ACCESS_UNTAGGED",
+                "ACCESS_TAGGED",
+                "ACCESS_UPLINK",
+                name="interfaceconfigtype",
+            ),
             nullable=False,
         ),
         sa.Column("data", postgresql.JSONB(astext_type=sa.Text())),
@@ -79,10 +100,9 @@ def upgrade():
         sa.Column("description", sa.String(length=255)),
         sa.ForeignKeyConstraint(["device_a_id"], ["device.id"]),
         sa.ForeignKeyConstraint(["device_b_id"], ["device.id"]),
-        sa.ForeignKeyConstraint(["site_id"], ["site.id"])
-
+        sa.ForeignKeyConstraint(["site_id"], ["site.id"]),
     )
-    
+
     op.create_table(
         "linknet",
         sa.Column("id", sa.Integer(), autoincrement=True, nullable=False, primary_key=True),
@@ -112,4 +132,4 @@ def downgrade():
 
     sa.Enum(name="devicetype").drop(op.get_bind(), checkfirst=False)
     sa.Enum(name="devicestate").drop(op.get_bind(), checkfirst=False)
-    sa.Enum(name="interfaceconfigtype").drop(op.get_bind(), checkfirst=False)
+    sa.Enum(name="interfaceconfigtype").drop(op.get_bind(), checkfirst=False)

docker/postgres/Dockerfile

@@ -1,6 +1,6 @@
 FROM postgres:11
 
-ARG SQLFILE=no-file[t] 
+ARG SQLFILE=no-file[t]
 
 COPY --chown=postgres:postgres ${SQLFILE} /docker-entrypoint-initdb.d/
 

src/cnaas_nms/db/git.py

@@ -1,5 +1,6 @@
 import datetime
 import enum
+import json
 import os
 import shutil
 from typing import Dict, Optional, Set, Tuple
@@ -23,6 +24,7 @@
 )
 from cnaas_nms.devicehandler.sync_history import add_sync_event
 from cnaas_nms.scheduler.thread_data import set_thread_data
+from cnaas_nms.tools.event import add_event
 from cnaas_nms.tools.log import get_logger
 from git import InvalidGitRepositoryError, Repo
 from git.exc import GitCommandError, NoSuchPathError
@@ -75,15 +77,23 @@ def refresh_repo(repo_type: RepoType = RepoType.TEMPLATES, scheduled_by: str = N
     # while another task is building configuration for devices using repo data
     with sqla_session() as session:
         job = Job()
-        job.start_job(function_name="refresh_repo", scheduled_by=scheduled_by)
         session.add(job)
         session.flush()
+        job.start_job(function_name="refresh_repo", scheduled_by=scheduled_by)
+        session.flush()
         job_id = job.id
         set_thread_data(job_id)
         logger = get_logger()
 
         logger.info("Trying to acquire lock for devices to run refresh repo")
         if not Joblock.acquire_lock(session, name="devices", job_id=job_id):
+            job.status = JobStatus.ABORTED
+            try:
+                event_data = {"job_id": job.id, "status": job.status.name}
+                json_data = json.dumps(event_data)
+                add_event(json_data=json_data, event_type="update", update_type="job")
+            except Exception:  # noqa: S110
+                pass
             raise JoblockError("Unable to acquire lock for configuring devices")
         try:
             result = _refresh_repo_task(repo_type, job_id=job_id)
@@ -95,6 +105,12 @@ def refresh_repo(repo_type: RepoType = RepoType.TEMPLATES, scheduled_by: str = N
                 Joblock.release_lock(session, job_id=job_id)
             except Exception:
                 logger.error("Unable to release devices lock after refresh repo job")
+            try:
+                event_data = {"job_id": job.id, "status": job.status.name}
+                json_data = json.dumps(event_data)
+                add_event(json_data=json_data, event_type="update", update_type="job")
+            except Exception:  # noqa: S110
+                pass
             return result
         except Exception as e:
             logger.exception("Exception while scheduling job for refresh repo")
@@ -106,6 +122,12 @@ def refresh_repo(repo_type: RepoType = RepoType.TEMPLATES, scheduled_by: str = N
                 Joblock.release_lock(session, job_id=job_id)
             except Exception:
                 logger.error("Unable to release devices lock after refresh repo job")
+            try:
+                event_data = {"job_id": job.id, "status": job.status.name}
+                json_data = json.dumps(event_data)
+                add_event(json_data=json_data, event_type="update", update_type="job")
+            except Exception:  # noqa: S110
+                pass
             raise e
 
 

src/cnaas_nms/db/joblock.py

@@ -93,8 +93,7 @@ def clear_locks(cls, session: sqla_session):
         try:
             return session.query(Joblock).delete()
         except DBAPIError as e:
-            if e.orig.pgcode == '42P01':
+            if e.orig.pgcode == "42P01":
                 raise JoblockError("Jobblock table doesn't exist yet, we assume it will be created soon.")
             else:
                 raise
-