[tooling] ruff: pyupgrade

build/dev/extract_json_field.py

@@ -3,7 +3,7 @@
 import json
 import sys
 
-with open(sys.argv[2], "r") as f:
+with open(sys.argv[2]) as f:
     try:
         obj = json.load(f)
     except ValueError as e:

monitoring/deployment_manager/actions/dss/v1/crdb.py

@@ -57,9 +57,7 @@ def print_ca_public_certs(context: Context):
         for i in range(len(certs))
         if public_key_bytes == _public_key_bytes(certs[i].public_key())
     ]
-    match_words = ["first", "second", "third"] + [
-        "{}th".format(i) for i in range(4, 50)
-    ]
+    match_words = ["first", "second", "third"] + [f"{i}th" for i in range(4, 50)]
     if not matches:
         context.log.warn(
             "This DSS instance's public key does not appear in any of the listed certificates"
@@ -86,7 +84,7 @@ def crdb_status(context: Context):
     )
     cluster = ClusterAPI(
         pod_session,
-        base_url="https://{}/api/v2".format(host_port),
+        base_url=f"https://{host_port}/api/v2",
         username=username,
         password=password,
     )
@@ -101,9 +99,9 @@ def crdb_status(context: Context):
         for n in nodes:
             k, v = n.summarize()
             summary[k] = v
-        context.log.msg("{} reports:\n".format(source) + yaml.dump(summary))
+        context.log.msg(f"{source} reports:\n" + yaml.dump(summary))
     else:
-        context.log.msg("{} not ready to query nodes".format(source))
+        context.log.msg(f"{source} not ready to query nodes")
 
 
 @deployment_action("dss/crdb/print_monitoring_user")
@@ -119,5 +117,5 @@ def print_monitoring_user(context: Context):
     username, password = crdb_sql.get_monitoring_user(
         context.clients.core, context.spec.dss.v1.namespace, context.spec.cluster.name
     )
-    context.log.msg("Username: {} Password: {}".format(username, password))
+    context.log.msg(f"Username: {username} Password: {password}")
     return

monitoring/deployment_manager/actions/k8s.py

@@ -7,7 +7,7 @@ def list_pods(context: Context):
     ret = context.clients.core.list_pod_for_all_namespaces(watch=False)
     msg = "\n".join(
         [
-            "{}\t{}\t{}".format(i.status.pod_ip, i.metadata.namespace, i.metadata.name)
+            f"{i.status.pod_ip}\t{i.metadata.namespace}\t{i.metadata.name}"
             for i in ret.items
         ]
     )
@@ -20,7 +20,7 @@ def list_ingress_controllers(context: Context):
     class_list = context.clients.networking.list_ingress_class()
     msg = "\n".join(
         [
-            "{}\t{}\t{}".format(c.metadata.name, c.spec.controller, c.spec.parameters)
+            f"{c.metadata.name}\t{c.spec.controller}\t{c.spec.parameters}"
             for c in class_list.items
         ]
     )

monitoring/deployment_manager/actions/test/hello_world.py

@@ -28,9 +28,7 @@ def destroy(context: Context) -> None:
     )
     if namespace is None:
         context.log.warn(
-            "Namespace `{}` does not exist in `{}` cluster".format(
-                context.spec.test.v1.namespace, context.spec.cluster.name
-            )
+            f"Namespace `{context.spec.test.v1.namespace}` does not exist in `{context.spec.cluster.name}` cluster"
         )
         return
 
@@ -41,9 +39,7 @@ def destroy(context: Context) -> None:
     )
 
     context.log.warn(
-        "Destroying hello_world system in `{}` namespace of `{}` cluster in 15 seconds...".format(
-            namespace.metadata.name, context.spec.cluster.name
-        )
+        f"Destroying hello_world system in `{namespace.metadata.name}` namespace of `{context.spec.cluster.name}` cluster in 15 seconds..."
     )
     sleep(15, "destruction of hello_world system may take a few seconds")
 

monitoring/deployment_manager/deploylib/common_k8s.py

@@ -1,4 +1,5 @@
-from typing import Any, Callable, Optional
+from collections.abc import Callable
+from typing import Any
 
 from structlog import BoundLogger
 
@@ -10,25 +11,23 @@ def get_resource(
     log: BoundLogger,
     resource_type: str,
     resource_name: str,
-) -> Optional[Any]:
-    log.msg("Checking for existing {}".format(resource_type), name=resource_name)
+) -> Any | None:
+    log.msg(f"Checking for existing {resource_type}", name=resource_name)
     resource_list = list_resources()
     matching_resources = [
         d for d in resource_list.items if d.metadata.name == resource_name
     ]
     if len(matching_resources) > 2:
         raise ValueError(
-            "Found {} {}s matching `{}`".format(
-                len(matching_resources), resource_type, resource_name
-            )
+            f"Found {len(matching_resources)} {resource_type}s matching `{resource_name}`"
         )
     if not matching_resources:
         return None
     return matching_resources[0]
 
 
 def upsert_resource(
-    existing_resource: Optional[Any],
+    existing_resource: Any | None,
     target_resource: Any,
     log: BoundLogger,
     resource_type: str,
@@ -38,16 +37,16 @@ def upsert_resource(
     if existing_resource is not None:
         if comparisons.specs_are_the_same(existing_resource, target_resource):
             log.msg(
-                "Existing {} does not need to be updated".format(resource_type),
+                f"Existing {resource_type} does not need to be updated",
                 name=existing_resource.metadata.name,
             )
             new_resource = existing_resource
         else:
-            log.msg("Updating existing {}".format(resource_type))
+            log.msg(f"Updating existing {resource_type}")
             new_resource = patch()
-            log.msg("Updated {}".format(resource_type), name=new_resource.metadata.name)
+            log.msg(f"Updated {resource_type}", name=new_resource.metadata.name)
     else:
-        log.msg("Creating new {}".format(resource_type))
+        log.msg(f"Creating new {resource_type}")
         new_resource = create()
-        log.msg("Created {}".format(resource_type), name=new_resource.metadata.name)
+        log.msg(f"Created {resource_type}", name=new_resource.metadata.name)
     return new_resource

monitoring/deployment_manager/deploylib/comparisons.py

@@ -1,4 +1,5 @@
-from typing import Any, Callable, Dict, List, Optional, Type
+from collections.abc import Callable
+from typing import Any
 
 from kubernetes.client import (
     V1Deployment,
@@ -9,11 +10,11 @@
     V1Service,
 )
 
-_special_comparisons: Dict[Type, Callable[[Any, Any], bool]] = {}
+_special_comparisons: dict[type, Callable[[Any, Any], bool]] = {}
 
 
 def specs_are_the_same(
-    obj1: Any, obj2: Any, field_paths: Optional[List[str]] = None
+    obj1: Any, obj2: Any, field_paths: list[str] | None = None
 ) -> bool:
     """Determine if the specifications for two Kubernetes objects are equivalent
 
@@ -48,13 +49,13 @@ def specs_are_the_same(
         else:
             return obj1 == obj2
 
-    sub_paths: Dict[str, Optional[List[str]]] = {}
+    sub_paths: dict[str, list[str] | None] = {}
     for field_path in field_paths:
         parts = field_path.split(".")
         if len(parts) == 1:
             if parts[0] in sub_paths:
                 raise ValueError(
-                    "Cannot compare {} and its subfield {}".format(parts[0], field_path)
+                    f"Cannot compare {parts[0]} and its subfield {field_path}"
                 )
             sub_paths[parts[0]] = None
         else:
@@ -76,7 +77,7 @@ def specs_are_the_same(
     return True
 
 
-def _special_comparison(type: Type):
+def _special_comparison(type: type):
     def decorator_declare_comparison(compare: Callable[[Any, Any], bool]):
         global _special_comparisons
         _special_comparisons[type] = compare

monitoring/deployment_manager/deploylib/crdb_cluster_api.py

@@ -1,5 +1,4 @@
 import math
-from typing import Dict, List, Optional, Tuple
 
 import arrow
 import requests
@@ -19,7 +18,7 @@ class ServerVersion(ImplicitDict):
 
 
 class Locality(ImplicitDict):
-    tiers: List[dict]
+    tiers: list[dict]
 
 
 class Node(ImplicitDict):
@@ -37,8 +36,8 @@ class Node(ImplicitDict):
     updated_at: int
     liveness_status: int
 
-    def summarize(self) -> Tuple[str, Dict[str, str]]:
-        key = "Node {} ({})".format(self.node_id, self.address.address_field)
+    def summarize(self) -> tuple[str, dict[str, str]]:
+        key = f"Node {self.node_id} ({self.address.address_field})"
         t0 = arrow.get(math.floor(self.started_at / 1e9))
         values = {
             "locality": " ".join(
@@ -51,7 +50,7 @@ def summarize(self) -> Tuple[str, Dict[str, str]]:
         return key, values
 
 
-class ClusterAPI(object):
+class ClusterAPI:
     """Wrapper for retrieving CockroachDB cluster information.
 
     API: https://www.cockroachlabs.com/docs/api/cluster/v2
@@ -61,8 +60,8 @@ def __init__(
         self,
         session: requests.Session,
         base_url: str = "https://localhost:8080/api/v2",
-        username: Optional[str] = None,
-        password: Optional[str] = None,
+        username: str | None = None,
+        password: str | None = None,
     ):
         self._session = session
         self._base_url = base_url
@@ -74,7 +73,7 @@ def __del__(self):
         self.log_out()
 
     def is_ready(self) -> bool:
-        resp = self._session.get("{}/health/?ready=true".format(self._base_url))
+        resp = self._session.get(f"{self._base_url}/health/?ready=true")
         if resp.status_code == 200:
             return True
         elif resp.status_code == 500:
@@ -87,7 +86,7 @@ def is_ready(self) -> bool:
             )
 
     def is_up(self) -> bool:
-        resp = self._session.get("{}/health/".format(self._base_url))
+        resp = self._session.get(f"{self._base_url}/health/")
         if resp.status_code == 200:
             return True
         elif resp.status_code == 500:
@@ -102,18 +101,14 @@ def is_up(self) -> bool:
     def log_in(self) -> str:
         if self._username is None:
             raise ValueError(
-                "Cannot log in to CockroachDB cluster at {} when username is not specified".format(
-                    self._base_url
-                )
+                f"Cannot log in to CockroachDB cluster at {self._base_url} when username is not specified"
             )
         if self._password is None:
             raise ValueError(
-                "Cannot log in to CockroachDB cluster at {} when password is not specified".format(
-                    self._base_url
-                )
+                f"Cannot log in to CockroachDB cluster at {self._base_url} when password is not specified"
             )
         resp = self._session.post(
-            "{}/login/".format(self._base_url),
+            f"{self._base_url}/login/",
             data={"username": self._username, "password": self._password},
         )
         resp.raise_for_status()
@@ -127,7 +122,7 @@ def log_in(self) -> str:
         self._session_auth = session
         return session
 
-    def _get_headers(self) -> Dict[str, str]:
+    def _get_headers(self) -> dict[str, str]:
         if self._session_auth is None:
             self.log_in()
         return {"X-Cockroach-API-Session": self._session_auth}
@@ -136,14 +131,14 @@ def log_out(self) -> None:
         if self._session_auth is None:
             return
         resp = self._session.post(
-            "{}/logout/".format(self._base_url), headers=self._get_headers()
+            f"{self._base_url}/logout/", headers=self._get_headers()
         )
         resp.raise_for_status()
         self._session_auth = None
 
-    def get_nodes(self) -> List[Node]:
+    def get_nodes(self) -> list[Node]:
         resp = self._session.get(
-            "{}/nodes/".format(self._base_url), headers=self._get_headers()
+            f"{self._base_url}/nodes/", headers=self._get_headers()
         )
         resp.raise_for_status()
         nodes = resp.json().get("nodes", None)

monitoring/deployment_manager/deploylib/crdb_sql.py

@@ -2,20 +2,19 @@
 import hashlib
 import random
 from dataclasses import dataclass
-from typing import List, Tuple
 
 import kubernetes.stream
 from kubernetes import client as k8s
 
 
 @dataclass
-class User(object):
+class User:
     username: str
-    options: List[str]
-    member_of: List[str]
+    options: list[str]
+    member_of: list[str]
 
 
-def execute_sql(client: k8s.CoreV1Api, namespace: str, sql_commands: List[str]) -> str:
+def execute_sql(client: k8s.CoreV1Api, namespace: str, sql_commands: list[str]) -> str:
     """Execute the specfied sql_commands directly on a CRDB node.
 
     :param client: Kubernetes core client which can access the CRDB node
@@ -40,7 +39,7 @@ def execute_sql(client: k8s.CoreV1Api, namespace: str, sql_commands: List[str])
     return resp
 
 
-def list_users(client: k8s.CoreV1Api, namespace: str) -> List[User]:
+def list_users(client: k8s.CoreV1Api, namespace: str) -> list[User]:
     lines = execute_sql(client, namespace, ["SHOW USERS"]).split("\n")
     lines = [line for line in lines[1:] if line]
     users = []
@@ -56,7 +55,7 @@ def list_users(client: k8s.CoreV1Api, namespace: str) -> List[User]:
 
 def get_monitoring_user(
     client: k8s.CoreV1Api, namespace: str, cluster_name: str
-) -> Tuple[str, str]:
+) -> tuple[str, str]:
     """Get the username and password for a CRDB user intended for monitoring.
 
     Whenever this routine is called, it sets the validity of the user to 1-2
@@ -92,11 +91,9 @@ def get_monitoring_user(
         client,
         namespace,
         [
-            "CREATE USER IF NOT EXISTS {}".format(username),
-            "GRANT admin TO {}".format(username),
-            "ALTER USER {} WITH LOGIN PASSWORD '{}' VALID UNTIL '{}'".format(
-                username, password, valid_until
-            ),
+            f"CREATE USER IF NOT EXISTS {username}",
+            f"GRANT admin TO {username}",
+            f"ALTER USER {username} WITH LOGIN PASSWORD '{password}' VALID UNTIL '{valid_until}'",
         ],
     )
 

monitoring/deployment_manager/deploylib/deployments.py

@@ -1,5 +1,3 @@
-from typing import Optional
-
 from kubernetes.client import AppsV1Api, V1Deployment, V1Namespace
 from structlog import BoundLogger
 
@@ -8,7 +6,7 @@
 
 def get(
     client: AppsV1Api, log: BoundLogger, namespace: V1Namespace, dep: V1Deployment
-) -> Optional[V1Deployment]:
+) -> V1Deployment | None:
     return common_k8s.get_resource(
         lambda: client.list_namespaced_deployment(namespace=namespace.metadata.name),
         log,

monitoring/deployment_manager/deploylib/ingresses.py

@@ -1,5 +1,3 @@
-from typing import Optional
-
 from kubernetes.client import NetworkingV1Api, V1Ingress, V1Namespace
 from structlog import BoundLogger
 
@@ -11,7 +9,7 @@ def get(
     log: BoundLogger,
     namespace: V1Namespace,
     ingress: V1Ingress,
-) -> Optional[V1Ingress]:
+) -> V1Ingress | None:
     return common_k8s.get_resource(
         lambda: client.list_namespaced_ingress(namespace=namespace.metadata.name),
         log,