Improve deploy performance for very large models (Issue #7262, PR #7278)

# For Reviewers

I changed some things, I would like to have reviewer input on what is acceptable:
1. more fine grained triggering of agents on put_partial: only notify agents that are in the increment
2. `put_version` and `put_partial` no longer wait for auto deploy to be completed **this may break wait conditions in tests everywhere**
3. increment cache is pre sorted per agent (slower on release (done once), faster for every agent (done often))
4. increment cache now refuses to move back to older versions
5. micro optimizations to use the DB more efficiently

** Do I need more tests anywhere? **

close #7262

# Self Check:

Strike through any lines that are not applicable (`~~line~~`) then check the box

- [x] Attached issue to pull request
- [x] Changelog entry
- [x] Type annotations are present
- [x] Code is clear and sufficiently documented
- [ ] No (preventable) type errors (check using make mypy or make mypy-diff)
```

src/inmanta/config.py:219: error: Argument 1 to "CronTab" has incompatible type "str | int"; expected "str"  [arg-type]
src/inmanta/config.py:311: error: Incompatible default for argument "validator" (default has type "Callable[[str], str]", argument has type "Callable[[str | T], T]")  [assignment]
src/inmanta/data/__init__.py:4982: error: Argument 1 to "loads" has incompatible type "object"; expected "str | bytes | bytearray"  [arg-type]
src/inmanta/data/__init__.py:5441: error: Signature of "get_list" incompatible with supertype "BaseDocument"  [override]
src/inmanta/data/__init__.py:5441: note:      Superclass:
src/inmanta/data/__init__.py:5441: note:          @classmethod
src/inmanta/data/__init__.py:5441: note:          def get_list(cls, *, order_by_column: str | None = ..., order: str | None = ..., limit: int | None = ..., offset: int | None = ..., no_obj: bool | None = ..., lock: RowLockMode | None = ..., connection: Connection | None = ..., **query: object) -> Coroutine[Any, Any, list[ConfigurationModel]]
src/inmanta/data/__init__.py:5441: note:      Subclass:
src/inmanta/data/__init__.py:5441: note:          @classmethod
src/inmanta/data/__init__.py:5441: note:          def get_list(cls, *, order_by_column: str | None = ..., order: str | None = ..., limit: int | None = ..., offset: int | None = ..., no_obj: bool | None = ..., lock: RowLockMode | None = ..., connection: Connection | None = ..., no_status: bool = ..., **query: object) -> Coroutine[Any, Any, list[ConfigurationModel]]
src/inmanta/data/__init__.py:5509: error: Argument 2 to "_get_status_field" of "ConfigurationModel" has incompatible type "object"; expected "str"  [arg-type]
src/inmanta/data/__init__.py:5513: error: Argument 1 to "append" of "list" has incompatible type "ConfigurationModel"; expected "dict[str, object]"  [arg-type]
src/inmanta/data/__init__.py:5514: error: Incompatible return value type (got "list[dict[str, object]]", expected "list[ConfigurationModel]")  [return-value]
src/inmanta/data/__init__.py:5796: error: "object" has no attribute "__iter__"; maybe "__dir__" or "__str__"? (not iterable)  [attr-defined]
src/inmanta/data/__init__.py:5886: error: "object" has no attribute "__iter__"; maybe "__dir__" or "__str__"? (not iterable)  [attr-defined]
src/inmanta/data/__init__.py:5962: error: Incompatible types in assignment (expression has type "object", variable has type "int")  [assignment]
src/inmanta/server/services/orchestrationservice.py:849: error: Argument 1 to "add_background_task" of "TaskHandler" has incompatible type "Coroutine[Any, Any, tuple[int, dict[str, Any] | None]]"; expected "Coroutine[object, None, Result | None]"  [arg-type]
src/inmanta/server/services/compilerservice.py:795: error: Incompatible types in assignment (expression has type "bool | int | float | str | dict[str, str | int | bool]", variable has type "int")  [assignment]
Generated HTML report (via XSLT): /home/wouter/projects/inmanta/mypy/index.html
```
- [x] Sufficient test cases (reproduces the bug/tests the requested feature)
- [x] Correct, in line with design
- [ ] End user documentation is included or an issue is created for end-user documentation (add ref to issue here: )
- [ ] If this PR fixes a race condition in the test suite, also push the fix to the relevant stable branche(s) (see [test-fixes](https://internal.inmanta.com/development/core/tasks/build-master.html#test-fixes) for more info)

# Preliminary results
(on 15k resources)

1. we cause a storm of agent pulls (each put_partial makes every agent pull, we recompile fast)
   - [x] make it smarter
3. increment calculation is both on agent pull path and the release version
   - very performance sensitive
   - [x] it pulled in all attributes, so large config, slow increment
   - [ ] it interferes with itself somehow or mucks up the cache (to be investigated)
       - [x] make cache invalidation monotonic (only ever allow newer versions)
   - [x] auto deploy triggering is done in-line: the compile has to wait for it, we could change that
4. pyinstrument works somewhat as an async profiler, but not to the point where the numbers add up
6. slow query log, lock timing log and all query log on postgresql help somehwat
7. we still have some slow queries, basically everything related to the whole version or the part that remains the same
 - ` UPDATE resource SET status='deployed' WHERE environment=$1 AND model=$2 AND resource_id =ANY($3) `
 -
 ```
 INSERT INTO resource(
	                environment,
	                model,
	                resource_id,
	                resource_type,
	                resource_id_value,
	                agent,
	                status,
	                attributes,
	                attribute_hash,
	                resource_set,
	                provides
	            )(
	                SELECT
	                    r.environment,
	                    $3,
	                    r.resource_id,
	                    r.resource_type,
	                    r.resource_id_value,
	                    r.agent,
	                    (
	                        CASE WHEN r.status='undefined'::resourcestate
	                        THEN 'undefined'::resourcestate
	                        ELSE 'available'::resourcestate
	                        END
	                    ) AS status,
	                    r.attributes AS attributes,
	                    r.attribute_hash,
	                    r.resource_set,
	                    r.provides
	                FROM resource AS r
	                WHERE r.environment=$1 AND r.model=$2 AND r.resource_set IS NOT NULL AND NOT r.resource_set=ANY($4)
	            )
	            RETURNING resource_id, resource_set
```
```
  SUM(CASE WHEN r.status NOT IN($1,$2) THEN 1 ELSE 0 END) AS done,
	                           to_json(array(SELECT jsonb_build_object('status', r2.status, 'id', r2.resource_id)
	                                         FROM resource AS r2
	                                         WHERE c.environment=r2.environment AND c.version=r2.model
	                                        )
	                           ) AS status
	                    FROM configurationmodel AS c LEFT OUTER JOIN resource AS r
	                    ON c.environment = r.environment AND c.version = r.model
	                    WHERE c.environment=$3
	                    GROUP BY c.environment, c.version

```
```
INSERT INTO public.resourceaction_resource (resource_id, resource_version, environment, resource_action_id) SELECT unnest($1::text[]), unnest($2::int[]), $3, $4
```

Current time taken over parts of put_partial

```
2024-02-28 17:41:57,308 performance              WARNING STARTING PUT PARTIAL
2024-02-28 17:41:57,312 performance              WARNING INPUT VALIDATION: 0.0035941700043622404
2024-02-28 17:41:57,441 performance              WARNING LOAD STAGE: 0.1291558850207366
2024-02-28 17:41:57,802 performance              WARNING MERGE STAGE: 0.3613146049901843
2024-02-28 17:41:59,651 performance              WARNING PUT STAGE: 1.849367157992674
2024-02-28 17:42:01,870 performance              WARNING AUTO DEPLOY STAGE: 2.218535807012813
```

Author: wouterdb

changelogs/unreleased/7262-performance.yml

@@ -0,0 +1,7 @@
+description: Improve deploy performance for very large models
+issue-nr: 7262
+change-type: minor
+destination-branches: [master, iso7]
+sections:
+  minor-improvement: "{{description}}"
+

src/inmanta/agent/config.py

@@ -91,7 +91,7 @@
     is_time,
 )
 
-agent_deploy_interval = Option(
+agent_deploy_interval: Option[int | str] = Option(
     "config",
     "agent-deploy-interval",
     0,

src/inmanta/config.py

@@ -133,19 +133,19 @@ def get(cls, section: Optional[str] = None, name: Optional[str] = None, default_
 
     @classmethod
     def get_for_option(cls, option: "Option[T]") -> T:
-        raw_value: Optional[str | T] = cls._get_value(option.section, option.name, option.get_default_value())
+        raw_value: str | T = cls._get_value(option.section, option.name, option.get_default_value())
         return option.validate(raw_value)
 
     @classmethod
-    def _get_value(cls, section: str, name: str, default_value: Optional[T] = None) -> Optional[str | T]:
+    def _get_value(cls, section: str, name: str, default_value: T) -> str | T:
         cfg: ConfigParser = cls.get_instance()
         val: Optional[str] = _get_from_env(section, name)
         if val is not None:
             LOGGER.debug(f"Setting {section}:{name} was set using an environment variable")
-        else:
-            val = cfg.get(section, name, fallback=default_value)
-
-        return val
+            return val
+        # Typing of this method in the sdk is not entirely accurate
+        # It just returns the fallback, whatever its type
+        return cfg.get(section, name, fallback=default_value)
 
     @classmethod
     def is_set(cls, section: str, name: str) -> bool:
@@ -205,12 +205,12 @@ def is_float(value: str) -> float:
     return float(value)
 
 
-def is_time(value: str) -> int:
+def is_time(value: str | int) -> int:
     """Time, the number of seconds represented as an integer value"""
     return int(value)
 
 
-def is_time_or_cron(value: str) -> Union[int, str]:
+def is_time_or_cron(value: str | int) -> Union[int, str]:
     """Time, the number of seconds represented as an integer value or a cron-like expression"""
     try:
         return is_time(value)
@@ -232,8 +232,10 @@ def is_bool(value: Union[bool, str]) -> bool:
     return boolean_states[value.lower()]
 
 
-def is_list(value: str) -> list[str]:
+def is_list(value: str | list[str]) -> list[str]:
     """List of comma-separated values"""
+    if isinstance(value, list):
+        return value
     return [] if value == "" else [x.strip() for x in value.split(",")]
 
 
@@ -304,9 +306,9 @@ def __init__(
         self,
         section: str,
         name: str,
-        default: Union[T, None, Callable[[], T]],
+        default: Union[T, Callable[[], T]],
         documentation: str,
-        validator: Callable[[Optional[str | T]], T] = is_str,
+        validator: Callable[[str | T], T] = is_str,
         predecessor_option: Optional["Option"] = None,
     ) -> None:
         self.section = section
@@ -342,10 +344,10 @@ def get_default_desc(self) -> str:
         else:
             return f"``{defa}``"
 
-    def validate(self, value: Optional[str | T]) -> T:
+    def validate(self, value: str | T) -> T:
         return self.validator(value)
 
-    def get_default_value(self) -> Optional[T]:
+    def get_default_value(self) -> T:
         defa = self.default
         if callable(defa):
             return defa()

src/inmanta/data/__init__.py

@@ -4621,6 +4621,40 @@ def convert_or_ignore(rvid):
         )
         return out
 
+    @classmethod
+    async def set_deployed_multi(
+        cls,
+        environment: uuid.UUID,
+        resource_ids: Sequence[m.ResourceIdStr],
+        version: int,
+        connection: Optional[asyncpg.connection.Connection] = None,
+    ) -> None:
+        query = "UPDATE resource SET status='deployed' WHERE environment=$1 AND model=$2 AND resource_id =ANY($3) "
+        async with cls.get_connection(connection) as connection:
+            await connection.execute(query, environment, version, resource_ids)
+
+    @classmethod
+    async def get_resource_ids_with_status(
+        cls,
+        environment: uuid.UUID,
+        resource_version_ids: list[m.ResourceIdStr],
+        version: int,
+        statuses: Sequence[const.ResourceState],
+        lock: Optional[RowLockMode] = None,
+        connection: Optional[asyncpg.connection.Connection] = None,
+    ) -> list[m.ResourceIdStr]:
+        query = (
+            "SELECT resource_id as resource_id FROM resource WHERE "
+            "environment=$1 AND model=$2 AND status = ANY($3) and resource_id =ANY($4) "
+        )
+        if lock:
+            query += lock.value
+        async with cls.get_connection(connection) as connection:
+            return [
+                m.ResourceIdStr(cast(str, r["resource_id"]))
+                for r in await connection.fetch(query, environment, version, statuses, resource_version_ids)
+            ]
+
     @classmethod
     async def get_undeployable(cls, environment: uuid.UUID, version: int) -> list["Resource"]:
         """
@@ -4794,29 +4828,42 @@ async def get_resources_for_version_raw_with_persistent_state(
         cls,
         environment: uuid.UUID,
         version: int,
-        projection: Optional[list[str]],
-        projection_presistent: Optional[list[str]],
+        projection: Optional[list[typing.LiteralString]],
+        projection_presistent: Optional[list[typing.LiteralString]],
+        project_attributes: Optional[list[typing.LiteralString]] = None,
         *,
         connection: Optional[Connection] = None,
     ) -> list[dict[str, object]]:
-        """This method performs none of the mangling required to produce valid resources!"""
+        """This method performs none of the mangling required to produce valid resources!
+
+        project_attributes performs a projection on the json attributes of the resources table
+
+        all projections must be disjoint, as they become named fields in the output record
+        """
 
         def collect_projection(projection: Optional[list[str]], prefix: str) -> str:
             if not projection:
                 return f"{prefix}.*"
             else:
                 return ",".join(f"{prefix}.{field}" for field in projection)
 
+        if project_attributes:
+            json_projection = "," + ",".join(f"r.attributes->'{v}' as {v}" for v in project_attributes)
+        else:
+            json_projection = ""
+
         query = f"""
-        SELECT {collect_projection(projection, 'r')}, {collect_projection(projection_presistent, 'ps')}
+        SELECT {collect_projection(projection, 'r')}, {collect_projection(projection_presistent, 'ps')} {json_projection}
             FROM {cls.table_name()} r JOIN resource_persistent_state ps ON r.resource_id = ps.resource_id
             WHERE r.environment=$1 AND ps.environment = $1 and r.model = $2;"""
 
         resource_records = await cls._fetch_query(query, environment, version, connection=connection)
         resources = [dict(record) for record in resource_records]
         for res in resources:
-            if "attributes" in res:
-                res["attributes"] = json.loads(res["attributes"])
+            if project_attributes:
+                for k in project_attributes:
+                    if res[k]:
+                        res[k] = json.loads(res[k])
         return resources
 
     @classmethod
@@ -5403,6 +5450,7 @@ async def get_list(
         no_obj: Optional[bool] = None,
         lock: Optional[RowLockMode] = None,
         connection: Optional[asyncpg.connection.Connection] = None,
+        no_status: bool = False,  # don't load the status field
         **query: object,
     ) -> list["ConfigurationModel"]:
         # sanitize and validate order parameters
@@ -5446,14 +5494,21 @@ async def get_list(
                     {lock_statement}"""
         query_result = await cls._fetch_query(query_string, *values, connection=connection)
         result = []
-        for record in query_result:
-            record = dict(record)
+        for in_record in query_result:
+            record = dict(in_record)
             if no_obj:
-                record["status"] = await cls._get_status_field(record["environment"], record["status"])
+                if no_status:
+                    record["status"] = {}
+                else:
+                    record["status"] = await cls._get_status_field(record["environment"], record["status"])
                 result.append(record)
             else:
                 done = record.pop("done")
-                status = await cls._get_status_field(record["environment"], record.pop("status"))
+                if no_status:
+                    status = {}
+                    record.pop("status")
+                else:
+                    status = await cls._get_status_field(record["environment"], record.pop("status"))
                 obj = cls(from_postgres=True, **record)
                 obj._done = done
                 obj._status = status
@@ -5703,23 +5758,23 @@ async def get_increment(
         deployed and different hash -> increment
         """
         # Depends on deploying
-        projection_a_resource = [
+        projection_a_resource: list[typing.LiteralString] = [
             "resource_id",
             "attribute_hash",
-            "attributes",
             "status",
         ]
-        projection_a_state = [
+        projection_a_state: list[typing.LiteralString] = [
             "last_success",
             "last_produced_events",
             "last_deployed_attribute_hash",
             "last_non_deploying_status",
         ]
-        projection = ["resource_id", "status", "attribute_hash"]
+        projection_a_attributes: list[typing.LiteralString] = ["requires", "send_event"]
+        projection: list[typing.LiteralString] = ["resource_id", "status", "attribute_hash"]
 
         # get resources for agent
         resources = await Resource.get_resources_for_version_raw_with_persistent_state(
-            environment, version, projection_a_resource, projection_a_state, connection=connection
+            environment, version, projection_a_resource, projection_a_state, projection_a_attributes, connection=connection
         )
 
         # to increment
@@ -5740,20 +5795,11 @@ async def get_increment(
                 continue
             # Now outstanding events
             last_success = resource["last_success"] or DATETIME_MIN_UTC
-            attributes = resource["attributes"]
-            assert isinstance(attributes, dict)  # mypy
-            for req in attributes["requires"]:
+            for req in resource["requires"]:
                 req_res = id_to_resource[req]
                 assert req_res is not None  # todo
-                req_res_attributes = req_res["attributes"]
-                assert isinstance(req_res_attributes, dict)  # mypy
                 last_produced_events = req_res["last_produced_events"]
-                if (
-                    last_produced_events is not None
-                    and last_produced_events > last_success
-                    and "send_event" in req_res_attributes
-                    and req_res_attributes["send_event"]
-                ):
+                if last_produced_events is not None and last_produced_events > last_success and req_res["send_event"]:
                     in_increment = True
                     break
 
@@ -5839,9 +5885,9 @@ async def get_increment(
 
         # build lookup tables
         for res in resources:
-            for req in res["attributes"]["requires"]:
+            for req in res["requires"]:
                 original_provides[req].append(res["resource_id"])
-            if "send_event" in res["attributes"] and res["attributes"]["send_event"]:
+            if res["send_event"]:
                 send_events.append(res["resource_id"])
 
         # recursively include stuff potentially receiving events from nodes in the increment

src/inmanta/server/agentmanager.py

@@ -1006,7 +1006,7 @@ async def _terminate_agents(self) -> None:
     async def _ensure_agents(
         self,
         env: data.Environment,
-        agents: list[str],
+        agents: Sequence[str],
         restart: bool = False,
         *,
         connection: Optional[asyncpg.connection.Connection] = None,

src/inmanta/server/config.py

@@ -244,18 +244,18 @@ def validate_fact_renew(value: object) -> int:
     "server", "purge-resource-action-logs-interval", 3600, "The number of seconds between resource-action log purging", is_time
 )
 
-server_resource_action_log_prefix = Option(
+server_resource_action_log_prefix: Option[str] = Option(
     "server",
     "resource_action_log_prefix",
     "resource-actions-",
     "File prefix in log-dir, containing the resource-action logs. The after the prefix the environment uuid and .log is added",
     is_str,
 )
 
-server_enabled_extensions = Option(
+server_enabled_extensions: Option[list[str]] = Option(
     "server",
     "enabled_extensions",
-    "",
+    list,
     "A list of extensions the server must load. Core is always loaded."
     "If an extension listed in this list is not available, the server will refuse to start.",
     is_list,
@@ -271,9 +271,9 @@ def validate_fact_renew(value: object) -> int:
 )
 
 
-def default_hangtime() -> str:
+def default_hangtime() -> int:
     """:inmanta.config:option:`server.agent-timeout` *3/4"""
-    return str(int(agent_timeout.get() * 3 / 4))
+    return int(agent_timeout.get() * 3 / 4)
 
 
 agent_hangtime = Option(