Retain safe validation, HTTP and Helm diagnostics in SDK logs (ae1690b0) · Commits · OOP / code / transformation-function-sdk

src/openop_tfsdk/_diagnostics.py

0 → 100644

+87 −0

Changes for src/openop_tfsdk/_diagnostics.py: 87 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""Small log-only summaries; never copy backend text, payloads or input values."""

		from typing import get_args

		import pydantic
		import requests
		from pydantic_core import ErrorType

		from openop_tfsdk.contract import execution_models

		_VALIDATION_RULES = frozenset(get_args(ErrorType))
		_HTTP_REASONS = {
		400: "bad_request",
		401: "authentication_required",
		403: "forbidden",
		404: "resource_not_found",
		408: "request_timeout",
		409: "conflict",
		422: "validation_failed",
		429: "rate_limited",
		500: "backend_internal_error",
		502: "bad_gateway",
		503: "service_unavailable",
		504: "gateway_timeout",
		}
		_HELM_REASONS = {
		"timed out waiting for the condition": "readiness_timeout",
		"context deadline exceeded": "deadline_exceeded",
		"kubernetes cluster unreachable": "cluster_unreachable",
		"unauthorized": "authentication_failed",
		"forbidden": "authorization_denied",
		"release: not found": "release_not_found",
		"cannot re-use a name": "release_name_in_use",
		}


		def validation_diagnostics(error, models=()):
		if not isinstance(error, pydantic.ValidationError):
		return {}
		fields = set()
		for model in (vars(execution_models).values(), models):
		if isinstance(model, type) and issubclass(model, pydantic.BaseModel):
		fields.update(model.model_fields)
		fields.update(
		field.alias for field in model.model_fields.values() if isinstance(field.alias, str)
		)
		errors = error.errors(include_url=False, include_context=False, include_input=False)
		summaries = []
		for item in errors[:5]:
		path = [
		part if isinstance(part, str) and part in fields else "?" for part in item["loc"][:6]
		]
		rule = item["type"]
		summaries.append(
		{
		"field": ".".join(path) or "$",
		"reason": rule if rule in _VALIDATION_RULES else "validation_error",
		}
		)
		return {"validation_errors": summaries, "validation_error_count": error.error_count()}


		def backend_diagnostics(error):
		# Existing clients wrap HTTP errors through cause/context; neither text nor body is needed.
		for current in (error, error.__cause__, error.__context__):
		if current is None:
		continue
		status = vars(current).get("status_code")
		if isinstance(current, requests.HTTPError) and current.response is not None:
		status = current.response.status_code
		if type(status) is int and 400 <= status <= 599:
		return {"backend_status": status, "reason": _HTTP_REASONS.get(status, "http_error")}
		if isinstance(current, requests.Timeout):
		return {"reason": "timeout"}
		if isinstance(current, requests.ConnectionError):
		return {"reason": "connection_failed"}
		return {}


		def helm_diagnostics(result):
		stderr = getattr(result, "stderr", None)
		text = stderr[:8192].lower() if isinstance(stderr, str) else ""
		reason = next(
		(reason for phrase, reason in _HELM_REASONS.items() if phrase in text),
		"helm_command_failed",
		)
		return {"exit_code": result.returncode, "reason": reason}

src/openop_tfsdk/contract/adapters/edgecloud.py

+6 −0

Changes for src/openop_tfsdk/contract/adapters/edgecloud.py: 6 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -14,6 +14,7 @@ from typing import Optional
		import pydantic
		import structlog

		from openop_tfsdk._diagnostics import backend_diagnostics, validation_diagnostics
		from openop_tfsdk.contract.errors import (
		DomainExecutionFailed,
		ResourceNotFound,
		@@ -28,6 +29,7 @@ from openop_tfsdk.contract.execution_models import (
		RuntimeState,
		)
		from openop_tfsdk.contract.models import SupportedAction
		from openop_tfsdk.edgecloud.adapters.i2edge import schemas
		from openop_tfsdk.edgecloud.adapters.i2edge.client import (
		EdgeApplicationManager as I2edgeClient,
		)
		@@ -130,6 +132,7 @@ class I2edgeCapabilityAdapter:
		method=method,
		error_code=ValidationError.error_code,
		error_type=type(error).__name__,
		**validation_diagnostics(error, vars(schemas).values()),
		)
		raise ValidationError(f"{method}: invalid request for i2edge") from error
		except I2EdgeError as error:
		@@ -140,6 +143,7 @@ class I2edgeCapabilityAdapter:
		method=method,
		error_code=ResourceNotFound.error_code,
		error_type=type(error).__name__,
		**backend_diagnostics(error),
		)
		raise ResourceNotFound(f"{method}: not found") from error
		log.error(
		@@ -147,6 +151,7 @@ class I2edgeCapabilityAdapter:
		method=method,
		error_code=DomainExecutionFailed.error_code,
		error_type=type(error).__name__,
		**backend_diagnostics(error),
		)
		raise DomainExecutionFailed(f"{method} failed") from error
		except Exception as error:
		@@ -155,5 +160,6 @@ class I2edgeCapabilityAdapter:
		method=method,
		error_code=DomainExecutionFailed.error_code,
		error_type=type(error).__name__,
		**backend_diagnostics(error),
		)
		raise DomainExecutionFailed(f"{method} failed") from error

src/openop_tfsdk/contract/adapters/kubernetes.py

+21 −4

Changes for src/openop_tfsdk/contract/adapters/kubernetes.py: 21 added lines, 4 removed lines.
Original line number	Diff line number	Diff line
		@@ -12,6 +12,7 @@ from typing import Optional

		import structlog

		from openop_tfsdk._diagnostics import helm_diagnostics
		from openop_tfsdk.contract.errors import DomainExecutionFailed, ValidationError
		from openop_tfsdk.contract.execution_models import (
		CapabilityLifecycleRequest,
		@@ -75,7 +76,11 @@ class KubernetesCapabilityAdapter:
		)

		if result.returncode != 0:
		log.error("helm_install_failed", error_code=DomainExecutionFailed.error_code)
		log.error(
		"helm_install_failed",
		error_code=DomainExecutionFailed.error_code,
		**helm_diagnostics(result),
		)
		raise DomainExecutionFailed("helm install failed")

		return CapabilityLifecycleResult(
		@@ -106,7 +111,11 @@ class KubernetesCapabilityAdapter:
		)

		if result.returncode != 0:
		log.warning("helm_status_failed", error_code=DomainExecutionFailed.error_code)
		log.warning(
		"helm_status_failed",
		error_code=DomainExecutionFailed.error_code,
		**helm_diagnostics(result),
		)
		return CapabilityRuntimeState(
		external_id=release_name,
		capability_kind=request.capability_kind,
		@@ -148,7 +157,11 @@ class KubernetesCapabilityAdapter:
		)

		if result.returncode != 0:
		log.error("helm_uninstall_failed", error_code=DomainExecutionFailed.error_code)
		log.error(
		"helm_uninstall_failed",
		error_code=DomainExecutionFailed.error_code,
		**helm_diagnostics(result),
		)
		raise DomainExecutionFailed("helm uninstall failed")

		return CapabilityLifecycleResult(
		@@ -182,7 +195,11 @@ class KubernetesCapabilityAdapter:
		)

		if result.returncode != 0:
		log.error("helm_upgrade_failed", error_code=DomainExecutionFailed.error_code)
		log.error(
		"helm_upgrade_failed",
		error_code=DomainExecutionFailed.error_code,
		**helm_diagnostics(result),
		)
		raise DomainExecutionFailed("helm upgrade failed")

		return CapabilityLifecycleResult(

src/openop_tfsdk/contract/adapters/network.py

+5 −0

Changes for src/openop_tfsdk/contract/adapters/network.py: 5 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -12,6 +12,7 @@ import pydantic
		import requests
		import structlog

		from openop_tfsdk._diagnostics import backend_diagnostics, validation_diagnostics
		from openop_tfsdk.contract.errors import (
		DomainExecutionFailed,
		ResourceNotFound,
		@@ -301,6 +302,7 @@ class NetworkCapabilityAdapter:
		"location_context_request_invalid",
		error_code=ValidationError.error_code,
		error_type=type(error).__name__,
		**validation_diagnostics(error, vars(schemas).values()),
		)
		raise ValidationError(
		f"location_context: invalid request for {self.adapter_type}"
		@@ -325,6 +327,7 @@ class NetworkCapabilityAdapter:
		method=method,
		error_code=ValidationError.error_code,
		error_type=type(error).__name__,
		**validation_diagnostics(error, vars(schemas).values()),
		)
		raise ValidationError(f"{method}: invalid request for {self.adapter_type}") from error
		except CoreHttpError as error:
		@@ -337,6 +340,7 @@ class NetworkCapabilityAdapter:
		method=method,
		error_code=DomainExecutionFailed.error_code,
		error_type=type(error).__name__,
		**backend_diagnostics(error),
		)
		raise DomainExecutionFailed(f"{method} failed") from error
		except Exception as error:
		@@ -345,6 +349,7 @@ class NetworkCapabilityAdapter:
		method=method,
		error_code=DomainExecutionFailed.error_code,
		error_type=type(error).__name__,
		**backend_diagnostics(error),
		)
		raise DomainExecutionFailed(f"{method} failed") from error

src/openop_tfsdk/edgecloud/adapters/kubernetes/client.py

+23 −3

Changes for src/openop_tfsdk/edgecloud/adapters/kubernetes/client.py: 23 added lines, 3 removed lines.
Original line number	Diff line number	Diff line
		@@ -2,6 +2,8 @@
		from typing import Dict, List, Optional

		from kubernetes.client import V1Deployment

		from openop_tfsdk._diagnostics import backend_diagnostics, validation_diagnostics
		from requests import Response

		from openop_tfsdk.edgecloud.adapters.kubernetes.lib.core.piedge_encoder import (
		@@ -117,7 +119,13 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		request=None,
		)
		except Exception as e:
		log.error("edge_cloud_call_failed", error_type=type(e).__name__)
		log.error(
		"edge_cloud_call_failed",
		method="get_all_onboarded_apps",
		error_type=type(e).__name__,
		**backend_diagnostics(e),
		**validation_diagnostics(e, vars(camara_schemas).values()),
		)
		status_code = 500
		return {
		"status": 500,
		@@ -260,7 +268,13 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		request=None,
		)
		except Exception as e:
		log.error("edge_cloud_call_failed", error_type=type(e).__name__)
		log.error(
		"edge_cloud_call_failed",
		method="get_all_deployed_apps",
		error_type=type(e).__name__,
		**backend_diagnostics(e),
		**validation_diagnostics(e, vars(camara_schemas).values()),
		)
		return {
		"status": 500,
		"code": "INTERNAL",
		@@ -343,7 +357,13 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		request=None,
		)
		except Exception as e:
		log.error("edge_cloud_call_failed", error_type=type(e).__name__)
		log.error(
		"edge_cloud_call_failed",
		method="undeploy_app",
		error_type=type(e).__name__,
		**backend_diagnostics(e),
		**validation_diagnostics(e, vars(camara_schemas).values()),
		)
		return {"status": 404, "code": "NOT_FOUND", "message": "Resource does not exist"}

		def get_edge_cloud_zones(