Chore: Fix pylint issues with new version of pylint

charms/worker/k8s/lib/charms/k8s/v0/k8sd_api_manager.py

@@ -54,12 +54,12 @@ class ErrorCodes(enum.Enum):
  """Enumerate the response codes from the k8s api.
 
  Attributes:
- StatusNodeUnavailable: returned when the node isn't in the cluster
- StatusNodeInUse: returned when the node is in the cluster already
+ STATUS_NODE_UNAVAILABLE: returned when the node isn't in the cluster
+ STATUS_NODE_IN_USE: returned when the node is in the cluster already
  """
 
- StatusNodeUnavailable = 520
- StatusNodeInUse = 521
+ STATUS_NODE_UNAVAILABLE = 520
+ STATUS_NODE_IN_USE = 521
 
 
 class K8sdAPIManagerError(Exception):
@@ -320,7 +320,7 @@ class UserFacingClusterConfig(BaseModel):
  cloud_provider: Optional[str] = Field(None, alias="cloud-provider")
 
 
-class UserFacingDatastoreConfig(BaseModel, allow_population_by_field_name=True): # type: ignore[call-arg]
+class UserFacingDatastoreConfig(BaseModel, allow_population_by_field_name=True):
  """Aggregated configuration model for the user-facing datastore aspects of a cluster.
 
  Attributes:

charms/worker/k8s/requirements.txt

@@ -3,7 +3,7 @@ charm-lib-interface-external-cloud-provider @ git+https://github.com/charmed-kub
 charm-lib-node-base @ git+https://github.com/charmed-kubernetes/layer-kubernetes-node-base@9b212854e768f13c26cc907bed51444e97e51b50#subdirectory=ops
 charm-lib-reconciler @ git+https://github.com/charmed-kubernetes/charm-lib-reconciler@f818cc30d1a22be43ffdfecf7fbd9c3fd2967502
 cosl==0.0.26
-ops==2.16.0
+ops==2.16.1
 pydantic==1.10.18
 PyYAML==6.0.2
 tomli == 2.0.1

charms/worker/k8s/scripts/update_alert_rules.py

@@ -52,11 +52,11 @@ def download_and_process_rule_files(temp_dir: Path):
  source_url = f"{SOURCE}/{file}"
  temp_file = temp_dir / file
  try:
- logging.info(f"Downloading {source_url}")
+ logging.info("Downloading %s", source_url)
  with urlopen(source_url) as response: # nosec
  process_rule_file(response, temp_file, source_url)
  except URLError as e:
- logging.error(f"Error fetching dashboard data: {e}")
+ logging.error("Error fetching dashboard data: %s", e)
 
 
 def process_rule_file(contents, destination_file: Path, source_url: str):
@@ -88,7 +88,7 @@ def process_rule_file(contents, destination_file: Path, source_url: str):
 
  with destination_file.open("w") as file:
  file.write("\n".join(data))
- logging.info(f"Processed and saved to {destination_file}")
+ logging.info("Processed and saved to %s", destination_file)
 
 
 def move_processed_files(temp_dir):
@@ -100,28 +100,25 @@ def move_processed_files(temp_dir):
  for temp_file in temp_dir.iterdir():
  final_path = ALERT_RULES_DIR / temp_file.name
  shutil.move(str(temp_file), str(final_path))
- logging.info(f"Moved {temp_file.name} to {final_path}")
+ logging.info("Moved %s to %s", temp_file.name, final_path)
 
 
 def apply_patches():
  """Apply patches to the downloaded and processed rule files."""
  for patch_file in PATCHES_DIR.glob("*"):
- logging.info(f"Applying patch {patch_file}")
+ logging.info("Applying patch %s", patch_file)
  subprocess.check_call(["/usr/bin/git", "apply", str(patch_file)])
 
 
 def main():
  """Fetch, process, and save AlertManager rules."""
  with TemporaryDirectory() as temp_dir:
  temp_path = Path(temp_dir)
- try:
- download_and_process_rule_files(temp_path)
- shutil.rmtree(ALERT_RULES_DIR, ignore_errors=True)
- ALERT_RULES_DIR.mkdir(parents=True)
- move_processed_files(temp_path)
- apply_patches()
- except Exception as e:
- logging.error("An error occurred: %s" % e)
+ download_and_process_rule_files(temp_path)
+ shutil.rmtree(ALERT_RULES_DIR, ignore_errors=True)
+ ALERT_RULES_DIR.mkdir(parents=True)
+ move_processed_files(temp_path)
+ apply_patches()
 
 
 if __name__ == "__main__":

charms/worker/k8s/scripts/update_dashboards.py

@@ -60,7 +60,7 @@ def fetch_dashboards(source_url: str):
  with urlopen(source_url) as response: # nosec
  return yaml.safe_load(response.read())
  except URLError as e:
- logging.error(f"Error fetching dashboard data: {e}")
+ logging.error("Error fetching dashboard data: %s", e)
  return None
 
 
@@ -101,17 +101,17 @@ def prepare_dashboard(json_value):
  return json.dumps(json_value, indent=4).replace("$datasource", "$prometheusds")
 
 
-def save_dashboard_to_file(name, data):
+def save_dashboard_to_file(name, data: str):
  """Save the prepared dashboard JSON to a file.
 
  Args:
  name (str): name of the dashboard file
  data (str): file content to write
  """
  filepath = os.path.join(TARGET_DIR, name)
- with open(filepath, "w") as f:
+ with open(filepath, "w", encoding="utf-8") as f:
  f.write(data)
- logging.info(f"Dashboard '{name}' saved to {filepath}")
+ logging.info("Dashboard '%s' saved to %s", name, filepath)
 
 
 def main():

charms/worker/k8s/src/charm.py

@@ -95,6 +95,7 @@ def _cluster_departing_unit(event: ops.EventBase) -> Union[Literal[False], ops.U
  isinstance(event, ops.RelationDepartedEvent)
  and event.relation.name in ["k8s-cluster", "cluster"]
  and event.departing_unit
+ or False
  )
 
 
@@ -131,7 +132,7 @@ def __init__(self, *args):
  self.labeller = LabelMaker(
  self, kubeconfig_path=self._internal_kubeconfig, kubectl=KUBECTL_PATH
  )
- self._stored.set_default(is_dying=False, cluster_name="")
+ self._stored.set_default(is_dying=False, cluster_name=str())
 
  self.cos_agent = COSAgentProvider(
  self,
@@ -206,8 +207,7 @@ def get_cluster_name(self) -> str:
  Returns:
  the cluster name.
  """
- unit, node = self.unit.name, self.get_node_name()
- if not self._stored.cluster_name:
+ if self._stored.cluster_name == "":
  if self.lead_control_plane and self.api_manager.is_cluster_bootstrapped():
  # TODO: replace with API call once available from the snap
  self._stored.cluster_name = str(uuid.uuid4())
@@ -221,8 +221,9 @@ def get_cluster_name(self) -> str:
  ):
  self._stored.cluster_name = self.collector.cluster_name(relation, True)
 
+ unit, node = self.unit.name, self.get_node_name()
  log.info("%s(%s) current cluster-name=%s", unit, node, self._stored.cluster_name)
- return self._stored.cluster_name
+ return str(self._stored.cluster_name)
 
  def get_node_name(self) -> str:
  """Return the lowercase hostname.
@@ -282,8 +283,8 @@ def _bootstrap_k8s_snap(self):
  bootstrap_config = BootstrapConfig()
  self._configure_datastore(bootstrap_config)
  self._configure_cloud_provider(bootstrap_config)
- bootstrap_config.service_cidr = self.config["service-cidr"]
- bootstrap_config.control_plane_taints = self.config["register-with-taints"].split()
+ bootstrap_config.service_cidr = str(self.config["service-cidr"])
+ bootstrap_config.control_plane_taints = str(self.config["register-with-taints"]).split()
  bootstrap_config.extra_sans = [_get_public_address()]
 
  status.add(ops.MaintenanceStatus("Bootstrapping Cluster"))
@@ -309,7 +310,7 @@ def _config_containerd_registries(self):
  registries, config = [], ""
  containerd_relation = self.model.get_relation("containerd")
  if self.is_control_plane:
- config = self.config["containerd_custom_registries"]
+ config = str(self.config["containerd_custom_registries"])
  registries = containerd.parse_registries(config)
  else:
  registries = containerd.recover(containerd_relation)
@@ -399,7 +400,7 @@ def _revoke_cluster_tokens(self, event: ops.EventBase):
  """
  log.info("Garbage collect cluster tokens")
  to_remove = None
- if self._stored.is_dying:
+ if self._stored.is_dying is True:
  to_remove = self.unit
  elif unit := _cluster_departing_unit(event):
  to_remove = unit
@@ -650,7 +651,7 @@ def _evaluate_removal(self, event: ops.EventBase) -> bool:
  Returns:
  True if being removed, otherwise False
  """
- if self._stored.is_dying:
+ if self._stored.is_dying is True:
  pass
  elif (unit := _cluster_departing_unit(event)) and unit == self.unit:
  # Juju says I am being removed
@@ -677,7 +678,7 @@ def _evaluate_removal(self, event: ops.EventBase) -> bool:
  elif isinstance(event, (ops.RemoveEvent, ops.StopEvent)):
  # If I myself am dying, its me!
  self._stored.is_dying = True
- return self._stored.is_dying
+ return bool(self._stored.is_dying)
 
  def _is_node_present(self, node: str = "") -> bool:
  """Determine if node is in the kubernetes cluster.

charms/worker/k8s/src/containerd.py

@@ -112,7 +112,7 @@ def __init__(self, *args, **kwargs):
  args: construction positional arguments
  kwargs: construction keyword arguments
  """
- super(Registry, self).__init__(*args, **kwargs)
+ super().__init__(*args, **kwargs)
  if not self.host and (host := self.url.host):
  self.host = host
 
@@ -177,11 +177,10 @@ def auth_config_header(self) -> Dict[str, Any]:
  log.debug("Configure basic auth for %s (%s)", self.url, self.host)
  v = self.username.get_secret_value() + ":" + self.password.get_secret_value()
  return {"Authorization": "Basic " + base64.b64encode(v.encode()).decode()}
- elif self.identitytoken:
+ if self.identitytoken:
  log.debug("Configure bearer token for %s (%s)", self.url, self.host)
  return {"Authorization": "Bearer " + self.identitytoken.get_secret_value()}
- else:
- return {}
+ return {}
 
  @property
  def hosts_toml(self) -> Dict[str, Any]:

charms/worker/k8s/src/snap.py

@@ -104,8 +104,7 @@ def _parse_management_arguments() -> List[SnapArgument]:
  if not revision.exists():
  raise snap_lib.SnapError(f"Failed to find file={revision}")
  try:
- with revision.open() as f:
- body = yaml.safe_load(f)
+ body = yaml.safe_load(revision.read_text(encoding="utf-8"))
  except yaml.YAMLError as e:
  log.error("Failed to load file=%s, %s", revision, e)
  raise snap_lib.SnapError(f"Failed to load file={revision}")
@@ -117,7 +116,9 @@ def _parse_management_arguments() -> List[SnapArgument]:
  raise snap_lib.SnapError(f"Failed to find revision for arch={arch}")
 
  try:
- args: List[SnapArgument] = [parse_obj_as(SnapArgument, arg) for arg in arch_spec] # type: ignore[arg-type]
+ args: List[SnapArgument] = [
+ parse_obj_as(SnapArgument, arg) for arg in arch_spec # type: ignore[arg-type]
+ ]
  except ValidationError as e:
  log.warning("Failed to validate args=%s (%s)", arch_spec, e)
  raise snap_lib.SnapError("Failed to validate snap args")