diff --git a/.gitignore b/.gitignore index e388e687..e9d37140 100644 --- a/.gitignore +++ b/.gitignore @@ -153,3 +153,4 @@ neo4j/Network-dev.graphml *.log* *.avsc CLAUDE.md +cc-usage-log.md diff --git a/Dockerfile-auth b/Dockerfile-auth index 54940136..5ccee5dd 100644 --- a/Dockerfile-auth +++ b/Dockerfile-auth @@ -1,7 +1,7 @@ -FROM python:3.11.0 +FROM python:3.13.13 MAINTAINER Komal Thareja -ARG HANDLERS_VER=1.9.1 +ARG HANDLERS_VER=2.0.0b9 RUN mkdir -p /usr/src/app WORKDIR /usr/src/app @@ -9,6 +9,9 @@ VOLUME ["/usr/src/app"] EXPOSE 11000 +RUN python3 -m venv /opt/venv +ENV PATH="/opt/venv/bin:$PATH" + RUN apt-get update RUN apt-get install cron -y RUN apt-get install sshpass -y @@ -22,12 +25,13 @@ COPY tools/audit.py /usr/src/app/ COPY tools/export.py /usr/src/app/ COPY tools/install.sh /usr/src/app/ -RUN pip3 install . +RUN pip install . RUN mkdir -p "/etc/fabric/message_bus/schema" RUN mkdir -p "/etc/fabric/actor/config" RUN mkdir -p "/var/log/actor" -RUN cp /usr/local/lib/python3.11/site-packages/fabric_mb/message_bus/schema/*.avsc /etc/fabric/message_bus/schema -RUN pip3 install fabric-am-handlers==${HANDLERS_VER} +RUN cp /opt/venv/lib/python3.13/site-packages/fabric_mb/message_bus/schema/*.avsc /etc/fabric/message_bus/schema +RUN pip install fabric-am-handlers==${HANDLERS_VER} +RUN ansible-galaxy collection install community.libvirt 'openstack.cloud:<2.0.0' cisco.nso RUN sh /usr/src/app/install.sh ENTRYPOINT ["/usr/src/app/docker-entrypoint.sh"] diff --git a/Dockerfile-broker b/Dockerfile-broker index 1d009858..08f8a4ef 100644 --- a/Dockerfile-broker +++ b/Dockerfile-broker @@ -1,4 +1,4 @@ -FROM python:3.11.0 +FROM python:3.13.13 MAINTAINER Komal Thareja RUN mkdir -p /usr/src/app @@ -7,6 +7,9 @@ VOLUME ["/usr/src/app"] EXPOSE 11000 +RUN python3 -m venv /opt/venv +ENV PATH="/opt/venv/bin:$PATH" + RUN apt-get update RUN apt-get install cron -y @@ -19,11 +22,11 @@ COPY tools/audit.py /usr/src/app/ COPY tools/export.py /usr/src/app/ COPY tools/install.sh /usr/src/app/ -RUN pip3 install . +RUN pip install . RUN mkdir -p "/etc/fabric/message_bus/schema" RUN mkdir -p "/etc/fabric/actor/config" RUN mkdir -p "/var/log/actor" -RUN cp /usr/local/lib/python3.11/site-packages/fabric_mb/message_bus/schema/*.avsc /etc/fabric/message_bus/schema +RUN cp /opt/venv/lib/python3.13/site-packages/fabric_mb/message_bus/schema/*.avsc /etc/fabric/message_bus/schema RUN sh /usr/src/app/install.sh ENTRYPOINT ["/usr/src/app/docker-entrypoint.sh"] diff --git a/Dockerfile-cf b/Dockerfile-cf index 445dd3fa..e30eecff 100644 --- a/Dockerfile-cf +++ b/Dockerfile-cf @@ -1,4 +1,4 @@ -FROM python:3.11.0 +FROM python:3.13.13 MAINTAINER Komal Thareja RUN mkdir -p /usr/src/app @@ -7,6 +7,9 @@ VOLUME ["/usr/src/app"] EXPOSE 11000 +RUN python3 -m venv /opt/venv +ENV PATH="/opt/venv/bin:$PATH" + RUN apt-get update RUN apt-get install cron -y @@ -19,14 +22,14 @@ COPY tools/export.py /usr/src/app/ COPY tools/export.py /usr/src/app/ COPY tools/install.sh /usr/src/app/ -RUN pip3 install . +RUN pip install . RUN mkdir -p "/etc/fabric/message_bus/schema" RUN mkdir -p "/etc/fabric/actor/config" RUN mkdir -p "/var/log/actor" -RUN cp /usr/local/lib/python3.11/site-packages/fabric_mb/message_bus/schema/*.avsc /etc/fabric/message_bus/schema +RUN cp /opt/venv/lib/python3.13/site-packages/fabric_mb/message_bus/schema/*.avsc /etc/fabric/message_bus/schema -RUN echo "0 2 * * * root /usr/local/bin/python3.11 /usr/src/app/audit.py -f /etc/fabric/actor/config/config.yaml -d 30 -c slices -o remove" >> /etc/crontab -RUN echo "0/15 * * * * root /usr/local/bin/python3.11 /usr/src/app/audit.py -f /etc/fabric/actor/config/config.yaml -d 30 -c slivers -o close" >> /etc/crontab +RUN echo "0 2 * * * root /opt/venv/bin/python /usr/src/app/audit.py -f /etc/fabric/actor/config/config.yaml -d 30 -c slices -o remove" >> /etc/crontab +RUN echo "0/15 * * * * root /opt/venv/bin/python /usr/src/app/audit.py -f /etc/fabric/actor/config/config.yaml -d 30 -c slivers -o close" >> /etc/crontab RUN service cron reload RUN service cron restart diff --git a/Dockerfile-orchestrator b/Dockerfile-orchestrator index 6e242750..9232c895 100644 --- a/Dockerfile-orchestrator +++ b/Dockerfile-orchestrator @@ -1,4 +1,4 @@ -FROM python:3.11.0 +FROM python:3.13.13 MAINTAINER Komal Thareja RUN mkdir -p /usr/src/app @@ -8,6 +8,9 @@ VOLUME ["/usr/src/app"] EXPOSE 11000 EXPOSE 8700 +RUN python3 -m venv /opt/venv +ENV PATH="/opt/venv/bin:$PATH" + RUN apt-get update RUN apt-get install cron -y @@ -20,11 +23,11 @@ COPY tools/audit.py /usr/src/app/ COPY tools/export.py /usr/src/app/ COPY tools/install.sh /usr/src/app/ -RUN pip3 install . +RUN pip install . RUN mkdir -p "/etc/fabric/message_bus/schema" RUN mkdir -p "/etc/fabric/actor/config" RUN mkdir -p "/var/log/actor" -RUN cp /usr/local/lib/python3.11/site-packages/fabric_mb/message_bus/schema/*.avsc /etc/fabric/message_bus/schema +RUN cp /opt/venv/lib/python3.13/site-packages/fabric_mb/message_bus/schema/*.avsc /etc/fabric/message_bus/schema RUN sh /usr/src/app/install.sh ENTRYPOINT ["/usr/src/app/docker-entrypoint.sh"] diff --git a/docker-entrypoint.sh b/docker-entrypoint.sh index ddd7b9d2..d057ddd3 100755 --- a/docker-entrypoint.sh +++ b/docker-entrypoint.sh @@ -1,3 +1,3 @@ #!/bin/sh service cron start -python3.11 -m $1 +/opt/venv/bin/python -m $1 diff --git a/fabric_cf/__init__.py b/fabric_cf/__init__.py index 98ee6bd4..f3a667d2 100644 --- a/fabric_cf/__init__.py +++ b/fabric_cf/__init__.py @@ -1,2 +1,2 @@ -__version__ = "1.10.0" +__version__ = "2.0.0b2" __VERSION__ = __version__ diff --git a/fabric_cf/actor/core/kernel/reservation_client.py b/fabric_cf/actor/core/kernel/reservation_client.py index 1b1ea08b..22fb72fb 100644 --- a/fabric_cf/actor/core/kernel/reservation_client.py +++ b/fabric_cf/actor/core/kernel/reservation_client.py @@ -759,6 +759,7 @@ def close(self, force: bool = False): self.logger.error(traceback.format_exc()) self.transition(prefix="close", state=ReservationStates.Closed, pending=ReservationPendingStates.None_) + finally: self.do_relinquish() else: self.transition_with_join(prefix="close", state=ReservationStates.Active, @@ -1272,8 +1273,10 @@ def probe_pending(self): # unreachable, it might be better to retry. self.transition(prefix=self.CLOSE_COMPLETE, state=ReservationStates.Closed, pending=ReservationPendingStates.None_) - # Note: the broker does not have information to ensure we - # are not cheating + finally: + # Always notify the Broker to release ticketed resources. + # do_relinquish() is idempotent (guarded by self.relinquished flag), + # so it is safe to call here even if update_lease() later calls it again. self.do_relinquish() def set_policy(self, *, policy: ABCClientPolicy): diff --git a/fabric_cf/actor/core/manage/kafka/kafka_broker.py b/fabric_cf/actor/core/manage/kafka/kafka_broker.py index 39f66936..391f5c1a 100644 --- a/fabric_cf/actor/core/manage/kafka/kafka_broker.py +++ b/fabric_cf/actor/core/manage/kafka/kafka_broker.py @@ -135,7 +135,7 @@ def extend_reservation(self, *, reservation: ID, new_end_time: datetime, sliver: request.auth = self.auth request.message_id = str(ID()) request.callback_topic = self.callback_topic - request.rid = str(reservation) + request.reservation_id = str(reservation) request.end_time = ActorClock.to_milliseconds(when=new_end_time) request.sliver = sliver diff --git a/fabric_cf/actor/core/manage/kafka/kafka_controller.py b/fabric_cf/actor/core/manage/kafka/kafka_controller.py index 6dbc0a02..310a1a96 100644 --- a/fabric_cf/actor/core/manage/kafka/kafka_controller.py +++ b/fabric_cf/actor/core/manage/kafka/kafka_controller.py @@ -41,11 +41,14 @@ from fim.slivers.base_sliver import BaseSliver from fim.user import GraphFormat +from fabric_mb.message_bus.messages.extend_reservation_avro import ExtendReservationAvro + from fabric_cf.actor.core.apis.abc_actor_mixin import ActorType from fabric_cf.actor.core.common.exceptions import ManageException from fabric_cf.actor.core.apis.abc_mgmt_controller_mixin import ABCMgmtControllerMixin from fabric_cf.actor.core.common.constants import Constants from fabric_cf.actor.core.manage.kafka.kafka_actor import KafkaActor +from fabric_cf.actor.core.time.actor_clock import ActorClock from fabric_cf.actor.core.util.id import ID @@ -80,6 +83,21 @@ def get_reservation_units(self, *, rid: ID, id_token: str = None) -> List[UnitAv if status.code == 0: return response.units + def extend_reservation(self, *, reservation: ID, new_end_time: datetime, sliver: BaseSliver, + dependencies: List[ReservationPredecessorAvro] = None) -> bool: + request = ExtendReservationAvro() + request.guid = str(self.management_id) + request.auth = self.auth + request.message_id = str(ID()) + request.callback_topic = self.callback_topic + request.reservation_id = str(reservation) + request.end_time = ActorClock.to_milliseconds(when=new_end_time) + request.sliver = sliver + + status, response = self.send_request(request) + + return status.code == 0 + def modify_reservation(self, *, rid: ID, modify_properties: dict) -> bool: raise ManageException(Constants.NOT_IMPLEMENTED) diff --git a/fabric_cf/actor/db/psql_database.py b/fabric_cf/actor/db/psql_database.py index 482a6691..9169203d 100644 --- a/fabric_cf/actor/db/psql_database.py +++ b/fabric_cf/actor/db/psql_database.py @@ -177,6 +177,7 @@ def get_actors(self) -> list: for row in session.query(Actors).all(): result.append(self.generate_dict_from_row(row=row)) except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -195,6 +196,7 @@ def get_actors_by_name_and_type(self, *, actor_name: str, act_type: int) -> list Actors.act_name.like(actor_name)).all(): result.append(self.generate_dict_from_row(row=row)) except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -211,6 +213,7 @@ def get_actors_by_name(self, *, act_name: str) -> list: for row in session.query(Actors).filter(Actors.act_name.like(act_name)).all(): result.append(self.generate_dict_from_row(row=row)) except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -234,6 +237,7 @@ def get_actor(self, *, name: str) -> dict: result = None self.logger.error("Actor: {} not found!".format(name)) except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -302,6 +306,7 @@ def get_miscellaneous(self, *, name: str) -> dict or None: else: return None except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -371,6 +376,7 @@ def get_manager_objects(self, *, act_id: int = None) -> list: for row in rows: result.append(self.generate_dict_from_row(row=row)) except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -387,6 +393,7 @@ def get_manager_objects_by_actor_name(self, *, act_name: str = None) -> list: act_id = act_obj['act_id'] result = self.get_manager_objects(act_id=act_id) except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -406,6 +413,7 @@ def get_manager_object(self, *, mo_key: str) -> dict: else: raise DatabaseException(self.OBJECT_NOT_FOUND.format("Manager Object", mo_key)) except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -421,6 +429,7 @@ def get_manager_containers(self) -> list: for mo_obj in session.query(ManagerObjects).filter(ManagerObjects.mo_act_id.is_(None)).all(): result.append(self.generate_dict_from_row(mo_obj)) except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -520,6 +529,7 @@ def get_slice_ids(self) -> list: for row in session.query(Slices).all(): result.append(row.slc_id) except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -572,6 +582,7 @@ def get_slice_count(self, *, project_id: str = None, email: str = None, states: return rows.count() except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e @@ -634,6 +645,7 @@ def get_slices(self, *, slice_id: str = None, slice_name: str = None, project_id for row in rows.all(): result.append(self.generate_dict_from_row(row=row)) except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -653,6 +665,7 @@ def get_slice_by_id(self, *, slc_id: int) -> dict: else: raise DatabaseException(self.OBJECT_NOT_FOUND.format("Slice", slc_id)) except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -923,6 +936,7 @@ def get_reservations(self, *, slice_id: str = None, graph_node_id: str = None, p for row in rows.all(): result.append(self.generate_dict_from_row(row=row)) except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -991,6 +1005,7 @@ def get_components(self, *, node_id: str, states: list[int], rsv_type: list[str] if row.bdf not in result[row.component]: result[row.component].append(row.bdf) except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -1046,6 +1061,7 @@ def get_links(self, *, node_id: str, states: list[int], rsv_type: list[str], result[row.node_id] = 0 result[row.node_id] += row.bw except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -1104,6 +1120,7 @@ def get_link_allocations(self, *, states: list[int], rsv_type: list[str], "site": row[4], }) except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -1162,6 +1179,7 @@ def get_component_allocations(self, *, states: list[int], "bdf": row[5], }) except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -1179,6 +1197,7 @@ def get_reservations_by_rids(self, *, rsv_resid_list: list) -> list: for row in session.query(Reservations).filter(Reservations.rsv_resid.in_(rsv_resid_list)).all(): result.append(self.generate_dict_from_row(row=row)) except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -1247,6 +1266,7 @@ def get_proxies(self, *, act_id: int) -> list: for row in session.query(Proxies).filter_by(prx_act_id=act_id).all(): result.append(self.generate_dict_from_row(row=row)) except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -1315,6 +1335,7 @@ def get_config_mappings(self, *, act_id: int) -> list: for row in session.query(ConfigMappings).filter_by(cfgm_act_id=act_id).all(): result.append(self.generate_dict_from_row(row=row)) except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -1403,6 +1424,7 @@ def get_client_by_guid(self, *, clt_guid: str) -> dict: raise DatabaseException(self.OBJECT_NOT_FOUND.format("Client", clt_guid)) result = self.generate_dict_from_row(clt_obj) except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -1419,6 +1441,7 @@ def get_clients(self) -> list: for row in session.query(Clients).all(): result.append(self.generate_dict_from_row(row=row)) except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -1469,6 +1492,7 @@ def get_unit(self, *, unt_uid: str) -> dict or None: return result result = self.generate_dict_from_row(unt_obj) except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -1490,6 +1514,7 @@ def get_units(self, *, rsv_resid: str): for row in session.query(Units).filter_by(unt_rsv_id=rsv_obj['rsv_id']).all(): result.append(self.generate_dict_from_row(row=row)) except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -1618,6 +1643,7 @@ def get_delegations(self, *, slc_guid: str = None, states: List[int] = None) -> for row in rows.all(): result.append(self.generate_dict_from_row(row=row)) except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -1637,6 +1663,7 @@ def get_delegation(self, *, dlg_graph_id: str) -> dict: else: raise DatabaseException(self.OBJECT_NOT_FOUND.format("Delegation", dlg_graph_id)) except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -1703,6 +1730,7 @@ def get_sites(self) -> list: for row in session.query(Sites).all(): result.append(self.generate_dict_from_row(row=row)) except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -1717,6 +1745,7 @@ def get_site(self, *, site_name: str) -> list: for row in session.query(Sites).filter_by(name=site_name).all(): result.append(self.generate_dict_from_row(row=row)) except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -1854,6 +1883,7 @@ def get_poas(self, *, poa_guid: str = None, project_id: str = None, email: str = for row in rows.all(): result.append(self.generate_dict_from_row(row=row)) except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e return result @@ -1908,6 +1938,7 @@ def get_metrics(self, *, project_id: str = None, user_id: str = None, excluded_p result.append(self.generate_dict_from_row(row=r)) return result except Exception as e: + session.rollback() self.logger.error(Constants.EXCEPTION_OCCURRED.format(e)) raise e diff --git a/fabric_cf/actor/fim/plugins/broker/aggregate_bqm_plugin.py b/fabric_cf/actor/fim/plugins/broker/aggregate_bqm_plugin.py index 46e0f41c..fd8f0e06 100644 --- a/fabric_cf/actor/fim/plugins/broker/aggregate_bqm_plugin.py +++ b/fabric_cf/actor/fim/plugins/broker/aggregate_bqm_plugin.py @@ -26,7 +26,7 @@ from __future__ import annotations import json -from datetime import datetime +from datetime import datetime, timezone from typing import Tuple, Dict, TYPE_CHECKING, List, Optional from collections import defaultdict @@ -167,6 +167,15 @@ def occupied_node_capacity(*, db: ABCDatabase, node_id: str, start: Optional[dat ReservationStates.Ticketed.value, ReservationStates.Nascent.value] + # When no time range is specified, default to "now" so that future + # advance reservations (Ticketed state) are not counted as currently + # occupied. Callers that need future availability (e.g. the calendar + # endpoint) pass explicit start/end and are unaffected. + if start is None and end is None: + now = datetime.now(timezone.utc) + start = now + end = now + # get existing reservations for this node existing_reservations = db.get_reservations(graph_node_id=node_id, states=states, start=start, end=end) # node capacities @@ -324,6 +333,19 @@ def plug_produce_bqm(self, *, cbm: ABCCBMPropertyGraph, **kwargs) -> ABCBQMPrope site_sliver.capacity_allocations = site_sliver.capacity_allocations + allocated_caps worker_sliver.capacity_allocations = allocated_caps + # Warn when allocations exceed capacity + w_cap = sliver.get_capacities() or Capacities() + if sliver.get_capacity_delegations() is not None: + _, dlg = sliver.get_capacity_delegations().get_sole_delegation() + if dlg.get_format() == DelegationFormat.SinglePool: + w_cap = dlg.get_details() + a_core = getattr(allocated_caps, 'core', 0) or 0 + c_core = getattr(w_cap, 'core', 0) or 0 + if a_core > c_core: + self.logger.warning( + f"Over-allocation detected on {sliver.get_name()} at {sliver.site}: " + f"cores_alloc={a_core} > cores_cap={c_core}") + # get the location if available if loc is None: loc = sliver.get_location() @@ -790,6 +812,15 @@ def plug_produce_bqm_summary(self, *, cbm: ABCCBMPropertyGraph, **kwargs) -> dic w_ram_alloc = getattr(worker_allocs, 'ram', 0) or 0 w_disk_alloc = getattr(worker_allocs, 'disk', 0) or 0 + # Warn when allocations exceed capacity (indicates stale or + # over-committed reservations on a host) + if w_core_alloc > w_core_cap or w_ram_alloc > w_ram_cap or w_disk_alloc > w_disk_cap: + self.logger.warning( + f"Over-allocation detected on {sliver.get_name()} at {s}: " + f"cores={w_core_alloc}/{w_core_cap} " + f"ram={w_ram_alloc}/{w_ram_cap} " + f"disk={w_disk_alloc}/{w_disk_cap}") + site_cores_cap += w_core_cap site_cores_alloc += w_core_alloc site_ram_cap += w_ram_cap @@ -810,23 +841,26 @@ def plug_produce_bqm_summary(self, *, cbm: ABCCBMPropertyGraph, **kwargs) -> dic # Worker-level components worker_components = {} if sliver.attached_components_info is not None: + # First pass: accumulate total capacity per type/model for comp in sliver.attached_components_info.list_devices(): rt = comp.resource_type rm = comp.resource_model comp_key = f"{rt}-{rm}" comp_cap = getattr(comp.capacities, 'unit', 0) or 0 - comp_alloc = 0 - if rt in allocated_comp_caps and rm in allocated_comp_caps[rt]: - comp_alloc = getattr(allocated_comp_caps[rt][rm], 'unit', 0) or 0 - if comp_key not in worker_components: worker_components[comp_key] = {"capacity": 0, "allocated": 0} worker_components[comp_key]["capacity"] += comp_cap - worker_components[comp_key]["allocated"] += comp_alloc - site_components[comp_key]["capacity"] += comp_cap - site_components[comp_key]["allocated"] += comp_alloc + + # Second pass: set allocations once per type/model from DB query results + for rt, models in allocated_comp_caps.items(): + for rm, alloc_cap in models.items(): + comp_key = f"{rt}-{rm}" + comp_alloc = getattr(alloc_cap, 'unit', 0) or 0 + if comp_key in worker_components: + worker_components[comp_key]["allocated"] += comp_alloc + site_components[comp_key]["allocated"] += comp_alloc # Build host record (level 2 includes per-host detail) if query_level == 2 or query_level == 0: diff --git a/fabric_cf/orchestrator/core/bqm_wrapper.py b/fabric_cf/orchestrator/core/bqm_wrapper.py index 8e5b545c..efecd7e0 100644 --- a/fabric_cf/orchestrator/core/bqm_wrapper.py +++ b/fabric_cf/orchestrator/core/bqm_wrapper.py @@ -34,7 +34,7 @@ class BqmWrapper: """ Implements cache for storing the BQM """ - def __init__(self): + def __init__(self, *, logger=None): self.graph_format = None self.bqm = None self.last_query_time = None @@ -46,8 +46,10 @@ def __init__(self): ''' self.refresh_interval_in_seconds = 2000 self.refresh_in_progress = False + self.refresh_started_at = None self.level = 1 self.graph_id = None + self.logger = logger def can_refresh(self) -> bool: """ @@ -55,9 +57,23 @@ def can_refresh(self) -> bool: @return True -> On first attempt or last query time is after refresh Interval; False otherwise """ current_time = datetime.now(timezone.utc) - if not self.refresh_in_progress and (self.last_query_time is None or - ((current_time - self.last_query_time).total_seconds() > - self.refresh_interval_in_seconds)): + + if self.refresh_in_progress: + if self.refresh_started_at and \ + (current_time - self.refresh_started_at).total_seconds() > self.refresh_interval_in_seconds: + if self.logger: + self.logger.warning( + f"BQM refresh has been stuck for " + f"{(current_time - self.refresh_started_at).total_seconds():.0f}s " + f"(timeout={self.refresh_interval_in_seconds}s); allowing new refresh" + ) + self.refresh_in_progress = False + self.refresh_started_at = None + else: + return False + + if self.last_query_time is None or \ + (current_time - self.last_query_time).total_seconds() > self.refresh_interval_in_seconds: return True return False @@ -72,10 +88,12 @@ def save(self, *, bqm: str, graph_format: GraphFormat, level: int): self.bqm = bqm self.last_query_time = datetime.now(timezone.utc) self.refresh_in_progress = False + self.refresh_started_at = None self.level = level def start_refresh(self): self.refresh_in_progress = True + self.refresh_started_at = datetime.now(timezone.utc) def get_bqm(self) -> str: """ diff --git a/fabric_cf/orchestrator/core/orchestrator_handler.py b/fabric_cf/orchestrator/core/orchestrator_handler.py index 5e479927..3a95dee4 100644 --- a/fabric_cf/orchestrator/core/orchestrator_handler.py +++ b/fabric_cf/orchestrator/core/orchestrator_handler.py @@ -151,10 +151,15 @@ def discover_broker_query_model(self, *, controller: ABCMgmtControllerMixin, tok if not start and not end and not includes and not excludes: saved_bqm = self.controller_state.get_saved_bqm(graph_format=graph_format, level=level) if saved_bqm is not None: - if not force_refresh and not saved_bqm.can_refresh() and not saved_bqm.refresh_in_progress: + if force_refresh: + saved_bqm.start_refresh() + elif saved_bqm.can_refresh(): + saved_bqm.start_refresh() + elif saved_bqm.refresh_in_progress and saved_bqm.get_bqm(): + # Serve stale cache while a refresh is already in progress broker_query_model = saved_bqm.get_bqm() else: - saved_bqm.start_refresh() + broker_query_model = saved_bqm.get_bqm() # Request the model from Broker as a fallback if not broker_query_model: @@ -243,10 +248,15 @@ def discover_broker_query_model_summary(self, *, controller: ABCMgmtControllerMi if not start and not end and not includes and not excludes: saved = self.controller_state.get_saved_summary(level=level) if saved is not None: - if not force_refresh and not saved.can_refresh() and not saved.refresh_in_progress: + if force_refresh: + saved.start_refresh() + elif saved.can_refresh(): + saved.start_refresh() + elif saved.refresh_in_progress and saved.get_bqm(): + # Serve stale cache while a refresh is already in progress summary_json = saved.get_bqm() else: - saved.start_refresh() + summary_json = saved.get_bqm() if not summary_json: broker = self.get_broker(controller=controller) diff --git a/fabric_cf/orchestrator/core/orchestrator_kernel.py b/fabric_cf/orchestrator/core/orchestrator_kernel.py index 232fc90b..db1ebc51 100644 --- a/fabric_cf/orchestrator/core/orchestrator_kernel.py +++ b/fabric_cf/orchestrator/core/orchestrator_kernel.py @@ -112,7 +112,7 @@ def save_bqm(self, *, bqm: str, graph_format: GraphFormat, level: int): key = f"{graph_format}-{level}" saved_bqm = self.bqm_cache.get(key) if saved_bqm is None: - saved_bqm = BqmWrapper() + saved_bqm = BqmWrapper(logger=self.get_logger()) saved_bqm.save(bqm=bqm, graph_format=graph_format, level=level) self.bqm_cache[key] = saved_bqm @@ -141,7 +141,7 @@ def save_summary(self, *, summary: str, level: int): key = f"SUMMARY-{level}" cached = self.summary_cache.get(key) if cached is None: - cached = BqmWrapper() + cached = BqmWrapper(logger=self.get_logger()) cached.save(bqm=summary, graph_format=GraphFormat.GRAPHML, level=level) self.summary_cache[key] = cached finally: diff --git a/pyproject.toml b/pyproject.toml index 253e8c62..697b237b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -15,7 +15,7 @@ dynamic = ["version"] keywords = ["Swagger", "Fabric Control Framework"] -requires-python = '>=3.9' +requires-python = '>=3.12' dependencies = [ "requests >= 2.28.1", "cryptography", @@ -26,11 +26,11 @@ dependencies = [ "connexion==2.14.2", "swagger-ui-bundle==0.0.9", "PyYAML", - "fabric_fss_utils==1.6.2", - "fabric-message-bus==1.10.0", + "fabric_fss_utils==1.7.0", + "fabric-message-bus==1.10.1", "fabric-fim==1.9.2", - "fabrictestbed==2.0.6", - "ansible" + "fabrictestbed==2.0.7", + "ansible==13.6.0" ] [project.optional-dependencies] diff --git a/tools/export.py b/tools/export.py index afcb8e22..41600921 100644 --- a/tools/export.py +++ b/tools/export.py @@ -40,6 +40,7 @@ from fabric_cf.actor.core.kernel.slice import SliceTypes from fabric_cf.actor.core.plugins.db.actor_database import ActorDatabase from fabric_cf.actor.core.container.globals import Globals, GlobalsSingleton +from fabric_cf.actor.core.apis.abc_controller_reservation import ABCControllerReservation from fabric_cf.actor.core.policy.inventory_for_type import InventoryForType from fabric_reports_client.reports_api import ReportsApi @@ -378,7 +379,11 @@ def export(self): for reservation in self.src_db.get_reservations(slice_id=slice_object.get_slice_id()): error_message = reservation.get_error_message() sliver_guid = str(reservation.get_reservation_id()) - sliver = InventoryForType.get_allocated_sliver(reservation=reservation) + if isinstance(reservation, ABCControllerReservation) and reservation.is_active() \ + and reservation.get_leased_resources() is not None: + sliver = reservation.get_leased_resources().get_sliver() + else: + sliver = InventoryForType.get_allocated_sliver(reservation=reservation) site_name = None host_name = None ip_subnet = None @@ -395,7 +400,7 @@ def export(self): site_name = sliver.get_site() if sliver.label_allocations and sliver.label_allocations.instance_parent: host_name = sliver.label_allocations.instance_parent - ip_subnet = str(sliver.management_ip) + ip_subnet = str(sliver.management_ip) if sliver.management_ip else None image = sliver.image_ref node_id = str(reservation.get_graph_node_id()) diff --git a/tools/install.sh b/tools/install.sh index 7490a2e5..d799f0df 100644 --- a/tools/install.sh +++ b/tools/install.sh @@ -1,8 +1,8 @@ #!/usr/bin/sh pip install fabric_reports_client -echo "0 * * * * root /usr/local/bin/python3.11 /usr/src/app/export.py --config_file /etc/fabric/actor/config/config.yaml" >> /etc/crontab -echo "0 * * * * root /usr/local/bin/python3.11 /usr/src/app/audit.py -f /etc/fabric/actor/config/config.yaml -a /etc/fabric/actor/config/vm_handler_config.yml -d 30 -c audit -o audit" >> /etc/crontab -#echo "0 2 * * * root /usr/local/bin/python3.11 /usr/src/app/audit.py -f /etc/fabric/actor/config/config.yaml -d 30 -c slices -o remove" >> /etc/crontab -#echo "*/15 * * * * root /usr/local/bin/python3.11 /usr/src/app/audit.py -f /etc/fabric/actor/config/config.yaml -c slivers -o close" >> /etc/crontab +echo "0 * * * * root /opt/venv/bin/python /usr/src/app/export.py --config_file /etc/fabric/actor/config/config.yaml" >> /etc/crontab +echo "0 * * * * root /opt/venv/bin/python /usr/src/app/audit.py -f /etc/fabric/actor/config/config.yaml -a /etc/fabric/actor/config/vm_handler_config.yml -d 30 -c audit -o audit" >> /etc/crontab +#echo "0 2 * * * root /opt/venv/bin/python /usr/src/app/audit.py -f /etc/fabric/actor/config/config.yaml -d 30 -c slices -o remove" >> /etc/crontab +#echo "*/15 * * * * root /opt/venv/bin/python /usr/src/app/audit.py -f /etc/fabric/actor/config/config.yaml -c slivers -o close" >> /etc/crontab service cron reload service cron restart \ No newline at end of file