diff --git a/README.md b/README.md index 168df55..89fc922 100644 --- a/README.md +++ b/README.md @@ -103,17 +103,29 @@ docker compose exec ground uv run flask user create ## Meter drivers -Thundercloud is not tied to any one meter driver. It talks to a driver over the HTTP+SSE -contract (and optionally gRPC) from the Thunder-Cloud 2.0 Open Source Meter Driver -Specification, so any compliant driver works — SparkNet-Http or a third party's. +A meter driver is a separate service that reaches the meters through their gateway radio; +Thundercloud talks to it over the HTTP+SSE contract (and optionally gRPC) of the +[Meter Driver Specification](https://github.com/EarthSpark/meter-driver-spec), version 1.4.0. +Any driver that implements the spec's required contract works, including the +meter-driver-emulator for development. Run the driver as its own service, then register it from the running ground app under **Global Settings > Meter Drivers > Register driver** by entering the base URL of its HTTP -service. Registered drivers become selectable per meter on the meter form. - -The groundbolt-dev workspace metarepo runs `sparknet-http` as part of its stack for -convenience during development; that is a choice of that stack, not a dependency of this -application. +service. Registration checks the driver's `openapi.json` against the spec and reports what is +missing. The document's `x-meter-driver` block lists the driver's interfaces; when it +advertises none, an `http` interface at the base URL is assumed. Selecting gRPC requires the +driver to advertise a gRPC target. Registration also asks the driver which init fields it +needs and writes them, with their types, to `meter_driver_configs/.json`. + +Fill in those fields under **Global Settings > Meter Drivers > Edit config** and save: the +values are validated against the discovered fields and sent to the driver's init endpoint, +and the outcome is recorded in the same file. The same init is re-sent whenever Thundercloud +starts. Registered drivers become selectable per meter on the meter form. + +The [groundbolt-dev workspace](https://github.com/EarthSpark/groundbolt-dev) runs a meter +driver alongside the webapp for development, the meter-driver-emulator, so developing +Thundercloud needs no gateway hardware and no vendor driver. Which driver a deployment uses is +the deployment's choice, not a dependency of this application. ## Development diff --git a/pyproject.toml b/pyproject.toml index 23cee37..e5822f6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -60,6 +60,10 @@ dependencies = [ "meter-driver-spec @ https://github.com/EarthSpark/meter-driver-spec/releases/download/v1.4.0/meter_driver_spec-1.4.0-py3-none-any.whl", "protobuf", "grpcio", + # sparkmeter.config.provider_settings catches pydantic.ValidationError + # raised by the meter-driver-spec models, so pydantic is a dependency of + # this package and not only of meter-driver-spec. + "pydantic", "click", # dates and times diff --git a/sparkmeter/config/configviews.py b/sparkmeter/config/configviews.py index 13f3ecf..62dad16 100644 --- a/sparkmeter/config/configviews.py +++ b/sparkmeter/config/configviews.py @@ -144,6 +144,7 @@ def meter_driver_edit(provider_id): provider_details = get_live_interface_details( provider["base_url"], selected_interface=provider["selected_interface"], + provider=provider, ) form = MeterDriverSettingsForm( @@ -171,6 +172,7 @@ def meter_driver_config(provider_id): provider_details = get_live_interface_details( provider["base_url"], selected_interface=provider["selected_interface"], + provider=provider, ) form = MeterDriverConfigEditorForm( formdata=request.form if request.method == "POST" else None, diff --git a/sparkmeter/config/provider_settings.py b/sparkmeter/config/provider_settings.py index 5e40262..73457fd 100644 --- a/sparkmeter/config/provider_settings.py +++ b/sparkmeter/config/provider_settings.py @@ -3,11 +3,14 @@ import json import logging +import re import uuid from pathlib import Path from urllib.parse import urlparse import httpx +from meter_driver_spec.http.models import RequirementsResponse +from pydantic import ValidationError from sparkmeter.config.configdomain import ConfigParameter from sparkmeter.config.configparameter import ParameterObject @@ -30,14 +33,60 @@ class DriverInitializationError(ValueError): _METER_DRIVER_CONFIG_DIR = _REPO_ROOT / "meter_driver_configs" logger = logging.getLogger(__name__) +# The operations the Meter Driver Specification (v1.4.0, docs/spec/index.md +# section 4) marks Required, other than GET /openapi.json, which is the +# document being checked. A driver's /openapi.json must document every one +# of these under its path with this method; the two Recommended routes +# (/v1/nodes/{node_id}/balance-and-flags, /v1/shutdown) are not demanded. +REQUIRED_CONTRACT_OPERATIONS = ( + ("get", "/v1/requirements"), + ("post", "/v1/init"), + ("post", "/v1/nodes/register"), + ("delete", "/v1/nodes/{node_id}"), + ("post", "/v1/nodes/{node_id}/configure-meter"), + ("post", "/v1/meters/configure"), + ("get", "/v1/events"), + ("get", "/v1/status"), + ("get", "/v1/healthz"), +) + +# The spec's interface-discovery extension block on /openapi.json (section 5.1). +DISCOVERY_EXTENSION = "x-meter-driver" + +# The gRPC profile's ConfigureDriver message (spec section 7, +# meter_driver.proto) has fixed fields, so a driver used over gRPC must list +# these on /v1/requirements whatever else it asks for. +GRPC_INIT_REQUIRED_FIELDS = ("heartbeat_period_duration", "aes_key") + +# A path template parameter, e.g. {node_id}; its name is not significant +# when matching a driver's paths against the spec's. +_PATH_PARAM_RE = re.compile(r"\{[^}]*\}") + +# The spec's AesKeyInput (section 3): a 16-byte key as 32 hex characters or +# as 16 byte values. +_AES_KEY_HEX_RE = re.compile(r"^[0-9a-fA-F]{32}$") +_AES_KEY_BYTES = 16 + +# Bound on nested schema composition (allOf parts, $ref chains) followed +# while reading a document, so a self-referential schema cannot recurse +# without end. +_SCHEMA_DEPTH_LIMIT = 8 + + +# --------------------------------------------------------------------------- +# OpenAPI schema helpers +# --------------------------------------------------------------------------- + def _resolve_local_ref(spec, ref): """Resolve a local JSON Pointer reference within an OpenAPI document.""" - if not ref or not ref.startswith("#/"): + if not isinstance(ref, str) or not ref.startswith("#/"): return None node = spec for part in ref[2:].split("/"): + if not isinstance(node, dict): + return None node = node.get(part.replace("~1", "/").replace("~0", "~")) if node is None: return None @@ -45,18 +94,83 @@ def _resolve_local_ref(spec, ref): def _resolve_schema(spec, schema): - """Resolve a schema object or local ref to a plain dict.""" - if not isinstance(schema, dict): - return {} - if "$ref" in schema: - return _resolve_local_ref(spec, schema["$ref"]) or {} - return schema + """Resolve a schema object to a plain dict, following $ref chains. + + Anything that is not a dict, or a $ref that does not resolve to one, + is an empty schema. A reference cycle stops at its first repeat. + """ + seen = set() + while isinstance(schema, dict) and "$ref" in schema: + ref = schema["$ref"] + if not isinstance(ref, str) or ref in seen: + return {} + seen.add(ref) + schema = _resolve_local_ref(spec, ref) + return schema if isinstance(schema, dict) else {} + + +def _schema_type(schema): + """Return a schema's JSON type name, or "" when it declares none. + + An OpenAPI 3.1 type array (["string", "null"]) yields its first non-null + entry. + """ + value = (schema or {}).get("type") + if isinstance(value, list): + for entry in value: + if entry != "null": + return str(entry).strip().lower() + return "" + return str(value or "").strip().lower() + + +def _object_schema(spec, schema, depth=0): + """Resolve an object schema, merging the properties and required lists of its allOf parts.""" + resolved = _resolve_schema(spec, schema) + parts = resolved.get("allOf") + if not isinstance(parts, list) or depth >= _SCHEMA_DEPTH_LIMIT: + return resolved + + merged = {key: value for key, value in resolved.items() if key != "allOf"} + properties = dict(merged["properties"]) if isinstance(merged.get("properties"), dict) else {} + required = list(merged["required"]) if isinstance(merged.get("required"), list) else [] + for part in parts: + part_schema = _object_schema(spec, part, depth + 1) + if isinstance(part_schema.get("properties"), dict): + properties.update(part_schema["properties"]) + if isinstance(part_schema.get("required"), list): + required.extend(name for name in part_schema["required"] if name not in required) + for key, value in part_schema.items(): + if key not in ("properties", "required", "allOf"): + merged.setdefault(key, value) + merged["properties"] = properties + merged["required"] = required + return merged + + +def _dig(node, *keys): + """Walk nested dicts by key; anything missing or not a dict yields {}.""" + for key in keys: + if not isinstance(node, dict): + return {} + node = node.get(key) + return node if isinstance(node, dict) else {} + + +# --------------------------------------------------------------------------- +# Optional extension: /v1/commands configure_provider vendor options +# +# Not part of the spec. A driver may additionally document a /v1/commands +# route whose configure_provider command carries a vendor_options object; +# its fields are offered as optional extras after the spec-discovered +# required fields. Nothing below is required for registration. +# --------------------------------------------------------------------------- def _command_type_values(spec, schema): """Extract the possible command_type discriminator values from a schema.""" resolved = _resolve_schema(spec, schema) - properties = resolved.get("properties") or {} + properties = resolved.get("properties") if isinstance(resolved.get("properties"), dict) else {} command_type = _resolve_schema(spec, properties.get("command_type") or {}) values = [] if "const" in command_type: @@ -68,21 +182,18 @@ def _command_type_values(spec, schema): def _find_configure_provider_schema(spec): """Locate the configure-provider command schema in the OpenAPI document.""" - request_schema = ( - (((spec.get("paths") or {}).get("/v1/commands") or {}).get("post") or {}) - .get("requestBody", {}) - .get("content", {}) - .get("application/json", {}) - .get("schema", {}) + request_schema = _dig( + spec, "paths", "/v1/commands", "post", "requestBody", "content", "application/json", "schema" ) request_schema = _resolve_schema(spec, request_schema) - for candidate in request_schema.get("oneOf") or []: + candidates = request_schema.get("oneOf") + for candidate in candidates if isinstance(candidates, list) else []: values = _command_type_values(spec, candidate) if "configure_provider" in values: return _resolve_schema(spec, candidate) - components = (spec.get("components") or {}).get("schemas") or {} + components = _dig(spec, "components", "schemas") for schema in components.values(): values = _command_type_values(spec, schema) if "configure_provider" in values: @@ -92,8 +203,8 @@ def _find_configure_provider_schema(spec): def _field_spec(name, schema, required): - """Normalize a vendor-option field description for the form layer.""" - field_type = str(schema.get("type") or "string").strip().lower() or "string" + """Normalize a field description for the form and config layers.""" + field_type = _schema_type(schema) or "string" return { "name": name, "label": str(schema.get("title") or name.replace("_", " ").title()), @@ -110,14 +221,21 @@ def _field_spec(name, schema, required): def _extract_vendor_option_fields(spec): """Extract configure-provider vendor option requirements from OpenAPI.""" command_schema = _find_configure_provider_schema(spec) - properties = command_schema.get("properties") or {} - vendor_schema = _resolve_schema(spec, properties.get("vendor_options") or {}) + properties = ( + command_schema.get("properties") if isinstance(command_schema.get("properties"), dict) else {} + ) + vendor_schema = _object_schema(spec, properties.get("vendor_options") or {}) fields = [] - required_fields = set(vendor_schema.get("required") or []) - for field_name, field_schema in (vendor_schema.get("properties") or {}).items(): - resolved_field = _resolve_schema(spec, field_schema) - fields.append(_field_spec(field_name, resolved_field, field_name in required_fields)) + required_fields = ( + set(vendor_schema["required"]) if isinstance(vendor_schema.get("required"), list) else set() + ) + vendor_properties = ( + vendor_schema.get("properties") if isinstance(vendor_schema.get("properties"), dict) else {} + ) + for field_name, field_schema in vendor_properties.items(): + resolved_field = _scalar_schema(spec, field_schema) + fields.append(_field_spec(str(field_name), resolved_field, field_name in required_fields)) return fields @@ -127,117 +245,154 @@ def _vendor_option_field_map(spec): return {field["name"]: field for field in _extract_vendor_option_fields(spec)} +# --------------------------------------------------------------------------- +# Init-field discovery: GET /v1/requirements typed by the InitRequest schema +# (spec sections 2, 5.2, 5.3 and 9) +# --------------------------------------------------------------------------- + + def _requirements_url(service_url): """Build the requirements endpoint URL from a service URL.""" return normalize_base_url(service_url) + "/v1/requirements" -def _fetch_requirements_payload(service_url, timeout=10.0): - """Fetch the optional driver requirements payload.""" - response = httpx.get(_requirements_url(service_url), timeout=timeout) - response.raise_for_status() - payload = response.json() - if not isinstance(payload, dict): - raise ProviderRegistrationError("driver requirements response must be a JSON object") - return payload +def _validation_summary(exc): + """One line naming each pydantic validation error's location and message.""" + return "; ".join( + "{}: {}".format(".".join(str(part) for part in error.get("loc") or ()) or "body", error.get("msg")) + for error in exc.errors() + ) -def _iter_candidate_requirement_schemas(spec): - """Yield object schemas that may describe driver requirement fields.""" - - def _walk(schema, depth=0): - if depth > 6: - return - resolved = _resolve_schema(spec, schema) - if not isinstance(resolved, dict): - return - properties = resolved.get("properties") or {} - if properties: - yield resolved - # Requirement fields are frequently nested (e.g. under - # vendor_options), so descend into object-typed properties too. - for prop_schema in properties.values(): - yield from _walk(prop_schema, depth + 1) - - components = (spec.get("components") or {}).get("schemas") or {} - for schema in components.values(): - yield from _walk(schema) +def _required_field_names_from_requirements(payload): + """Return the `required_fields` names from a RequirementsResponse, in the driver's order. - for path_item in (spec.get("paths") or {}).values(): - if not isinstance(path_item, dict): - continue - for operation in path_item.values(): - if not isinstance(operation, dict): - continue - schema = ( - ((operation.get("requestBody") or {}).get("content") or {}) - .get("application/json", {}) - .get("schema", {}) + The payload is validated as the spec's RequirementsResponse model + (`required_fields` is an array of strings). A blank name is rejected; + a repeated name is kept once. + """ + try: + requirements = RequirementsResponse.model_validate(payload) + except ValidationError as exc: + raise ProviderRegistrationError( + "driver requirements response is not a valid RequirementsResponse: {}".format( + _validation_summary(exc) + ) + ) from exc + names = [] + for name in requirements.required_fields: + if not name.strip(): + raise ProviderRegistrationError("driver requirements response lists a blank field name") + if name not in names: + names.append(name) + return names + + +def _fetch_requirements_payload(service_url, timeout=10.0): + """GET /v1/requirements and return the driver's required field names.""" + url = _requirements_url(service_url) + try: + response = httpx.get(url, timeout=timeout) + response.raise_for_status() + except httpx.HTTPError as exc: + raise ProviderRegistrationError( + "could not fetch driver requirements from /v1/requirements: {}".format( + str(exc) or exc.__class__.__name__ ) - yield from _walk(schema) + ) from exc + try: + payload = response.json() + except ValueError as exc: + raise ProviderRegistrationError("driver requirements response is not valid JSON") from exc + return _required_field_names_from_requirements(payload) + +def _init_request_schema(spec): + """Return the document's InitRequest schema as one object schema. -def _best_matching_requirements_schema(spec, required_fields): - """Find the schema whose properties best match the reported required fields.""" - required_names = set(required_fields or []) - best_schema = {} - best_score = 0 - for schema in _iter_candidate_requirement_schemas(spec): - properties = set((schema.get("properties") or {}).keys()) - score = len(required_names & properties) - if score > best_score: - best_schema = schema - best_score = score - return best_schema if best_score else {} + The POST /v1/init request-body schema is authoritative; a document that + documents init fields only under components.schemas.InitRequest is + read from there. $ref chains are followed and allOf parts merged. + """ + schema = _dig(spec, "paths", "/v1/init", "post", "requestBody", "content", "application/json", "schema") + resolved = _object_schema(spec, schema) + if isinstance(resolved.get("properties"), dict) and resolved["properties"]: + return resolved + return _object_schema(spec, _dig(spec, "components", "schemas").get("InitRequest") or {}) -# Type hints for the spec's standard driver init fields, used when the -# driver's OpenAPI does not describe a required field's schema itself. -_STANDARD_INIT_FIELD_TYPES = { - "aes_key": "string", - "channel": "integer", - "heartbeat_period_duration": "integer", -} +def _scalar_schema(spec, schema): + """Resolve a property schema to the alternative describing its scalar wire form. + + A oneOf/anyOf without its own type (the spec's AesKeyInput: 32-hex + string or 16-byte array) reduces to its first string-typed alternative, + else its first alternative, so the form layer sees one type and pattern. + """ + resolved = _resolve_schema(spec, schema) + alternatives = resolved.get("oneOf") or resolved.get("anyOf") + if _schema_type(resolved) or not isinstance(alternatives, list) or not alternatives: + return resolved + # Alternatives that are not schema objects (or $refs to none) are skipped. + resolved_alternatives = [ + resolved_alternative + for resolved_alternative in (_resolve_schema(spec, alternative) for alternative in alternatives) + if resolved_alternative + ] + for alternative in resolved_alternatives: + if _schema_type(alternative) == "string": + return alternative + return resolved_alternatives[0] if resolved_alternatives else {} def _extract_fields_from_requirements(spec, required_fields): - """Build normalized field specs from a requirements list plus OpenAPI schema hints.""" - schema = _best_matching_requirements_schema(spec, required_fields) - properties = schema.get("properties") or {} - schema_required = set(schema.get("required") or []) + """Build field specs for the advertised init fields, typed from InitRequest. + + A field the document does not describe is typed as a string, with a + warning: the spec (section 5.2) says the names must match the init + request schema in /openapi.json. + """ + init_schema = _init_request_schema(spec) + properties = init_schema.get("properties") if isinstance(init_schema.get("properties"), dict) else {} fields = [] for name in required_fields: - resolved = dict(_resolve_schema(spec, properties.get(name) or {})) - if not resolved.get("type") and name in _STANDARD_INIT_FIELD_TYPES: - resolved["type"] = _STANDARD_INIT_FIELD_TYPES[name] - fields.append( - _field_spec( + if name not in properties: + logger.warning( + "driver requirement field %r is not described by the contract's InitRequest schema; " + "treating it as a string", name, - resolved, - name in schema_required or name in set(required_fields), ) - ) + fields.append(_field_spec(name, _scalar_schema(spec, properties.get(name) or {}), True)) return fields def _extract_driver_requirement_fields(base_url, spec, timeout=10.0): - """Discover required driver fields from /v1/requirements, else OpenAPI.""" - vendor_option_fields = _extract_vendor_option_fields(spec) - if not vendor_option_fields: - # The driver declares no configurable requirements, so there is - # nothing to enrich and no reason to probe /v1/requirements. - return [] - try: - payload = _fetch_requirements_payload(base_url, timeout=timeout) - required_fields = payload.get("required_fields") or [] - if isinstance(required_fields, list) and required_fields: - normalized = [str(name).strip() for name in required_fields if str(name).strip()] - if normalized: - return _extract_fields_from_requirements(spec, normalized) - except (httpx.HTTPError, ValueError, ProviderRegistrationError): - pass - - return vendor_option_fields + """Discover the driver's init fields. + + GET /v1/requirements is required and its list order is kept; each name + is typed from the document's InitRequest schema. Every other + InitRequest property follows as an optional field, in schema order. + Fields from the optional /v1/commands vendor-option extension, if the + document has one, follow as optional extras for names not already + listed, so InitRequest typing wins for a name present in both. + """ + names = _fetch_requirements_payload(base_url, timeout=timeout) + fields = _extract_fields_from_requirements(spec, names) + known = {field["name"] for field in fields} + init_schema = _init_request_schema(spec) + init_properties = init_schema.get("properties") if isinstance(init_schema.get("properties"), dict) else {} + for name, schema in init_properties.items(): + if name in known: + continue + fields.append(_field_spec(name, _scalar_schema(spec, schema), False)) + known.add(name) + for vendor_field in _extract_vendor_option_fields(spec): + if vendor_field["name"] in known: + continue + extra = dict(vendor_field) + extra["required"] = False + fields.append(extra) + known.add(extra["name"]) + return fields def _get_parameter(name): @@ -491,20 +646,47 @@ def parse_provider_config_text(config_text): return payload -def _required_field_names(payload): - """Return the required field names from a config payload.""" - names = payload.get("required_fields") or [] - if not isinstance(names, list): +def _stored_field_specs(payload): + """Return a config payload's field specs keyed by name, in stored order. + + Entries in `required_fields` are the specs written at registration + (dicts with name, type, required, pattern, minimum, maximum, ...). A + bare name is a config written before field types were recorded: it is + treated as a required string field, and a warning says that + re-registering the driver records its types. + """ + entries = payload.get("required_fields") or [] + if not isinstance(entries, list): raise DriverConfigError("required_fields must be a list") - normalized = [] - for entry in names: + + specs = {} + untyped = [] + for entry in entries: if isinstance(entry, dict): name = str(entry.get("name") or "").strip() - else: - name = str(entry).strip() + if name: + specs[name] = entry + continue + name = str(entry).strip() if name: - normalized.append(name) - return normalized + specs[name] = {"name": name, "type": "string", "required": True} + untyped.append(name) + if untyped: + logger.warning( + "driver config lists required_fields without types (%s); they are treated as required " + "strings. Re-register the driver to record the types its contract declares.", + ", ".join(untyped), + ) + return specs + + +def _required_field_names(payload): + """Return the names a config payload must supply values for. + + A spec entry marked `"required": false` is an optional extra and is + not demanded. + """ + return [name for name, spec in _stored_field_specs(payload).items() if spec.get("required") is not False] def _field_values(payload): @@ -515,25 +697,39 @@ def _field_values(payload): return values -def _required_field_specs(payload): - """Return required field metadata keyed by field name.""" - names = payload.get("required_fields") or [] - if not isinstance(names, list): - raise DriverConfigError("required_fields must be a list") +def _is_blank(value): + """Whether a stored value counts as not supplied.""" + return value is None or (isinstance(value, str) and not value.strip()) - specs = {} - for entry in names: - if not isinstance(entry, dict): - continue - name = str(entry.get("name") or "").strip() - if name: - specs[name] = entry - return specs + +def _coerce_aes_key(value): + """Return an aes_key value in one of the spec's AesKeyInput forms. + + The wire forms (spec section 3) are 32 hex characters or an array of + exactly 16 integers 0..255. + """ + if isinstance(value, str): + value = value.strip() + if _AES_KEY_HEX_RE.fullmatch(value): + return value + elif ( + isinstance(value, list) + and len(value) == _AES_KEY_BYTES + and all(isinstance(item, int) and not isinstance(item, bool) and 0 <= item <= 255 for item in value) + ): + return list(value) + raise DriverConfigError( + "field 'aes_key' must be 32 hex characters or an array of {} byte values".format(_AES_KEY_BYTES) + ) def _coerce_field_value(name, value, spec): - """Coerce a raw JSON field value to the type required by the driver.""" - field_type = str((spec or {}).get("type") or "string").strip().lower() + """Coerce a raw JSON field value to the type the driver's contract declares.""" + if name == "aes_key": + return _coerce_aes_key(value) + if isinstance(value, str): + value = value.strip() + field_type = _schema_type(spec) or "string" if field_type == "integer": try: return int(value) @@ -553,23 +749,69 @@ def _coerce_field_value(name, value, spec): if normalized in ("false", "0", "no", "off"): return False raise DriverConfigError("field {!r} must be a boolean".format(name)) - return value + if field_type == "array": + if isinstance(value, list): + return list(value) + raise DriverConfigError("field {!r} must be an array".format(name)) + if field_type == "object": + if isinstance(value, dict): + return dict(value) + raise DriverConfigError("field {!r} must be an object".format(name)) + if isinstance(value, (list, dict)): + raise DriverConfigError("field {!r} must be a string".format(name)) + return value if isinstance(value, str) else str(value) + + +def _check_field_constraints(name, value, spec): + """Enforce the pattern, minimum and maximum the driver's contract declared for a field.""" + pattern = str((spec or {}).get("pattern") or "") + if pattern and isinstance(value, str): + try: + matched = re.search(pattern, value) is not None + except re.error: + logger.warning("field %r declares an unusable pattern %r; not checked", name, pattern) + matched = True + if not matched: + raise DriverConfigError("field {!r} must match the pattern {}".format(name, pattern)) + if isinstance(value, bool) or not isinstance(value, (int, float)): + return + minimum = (spec or {}).get("minimum") + if isinstance(minimum, (int, float)) and not isinstance(minimum, bool) and value < minimum: + raise DriverConfigError("field {!r} must be at least {}".format(name, minimum)) + maximum = (spec or {}).get("maximum") + if isinstance(maximum, (int, float)) and not isinstance(maximum, bool) and value > maximum: + raise DriverConfigError("field {!r} must be at most {}".format(name, maximum)) def validate_provider_config_payload(payload): - """Validate required field presence in a driver config payload.""" - required_fields = _required_field_names(payload) - required_field_specs = _required_field_specs(payload) + """Validate a driver config payload and return the typed init body. + + Every required field needs a non-blank value. Values are coerced to + the types the driver's contract declared and checked against its + pattern/minimum/maximum. Only fields the driver asked for are + returned: an optional field left blank is omitted, and a key the + driver never listed is dropped with a warning. + """ + specs = _stored_field_specs(payload) + required_fields = [name for name, spec in specs.items() if spec.get("required") is not False] field_values = _field_values(payload) - missing = [ - name for name in required_fields if name not in field_values or field_values[name] in (None, "") - ] + missing = [name for name in required_fields if _is_blank(field_values.get(name))] if missing: raise DriverConfigError("required fields are missing values: {}".format(", ".join(missing))) - coerced_field_values = { - name: _coerce_field_value(name, value, required_field_specs.get(name)) - for name, value in field_values.items() - } + + unknown = [name for name in field_values if name not in specs] + if unknown: + logger.warning( + "driver config field_values has keys the driver did not ask for; not sent: %s", ", ".join(unknown) + ) + + coerced_field_values = {} + for name, spec in specs.items(): + if name not in field_values or _is_blank(field_values[name]): + continue + value = _coerce_field_value(name, field_values[name], spec) + _check_field_constraints(name, value, spec) + coerced_field_values[name] = value return { "required_fields": required_fields, "field_values": coerced_field_values, @@ -687,14 +929,51 @@ def initialize_configured_providers_on_startup(timeout=10.0): return results -def save_provider_settings( - service_url, selected_interface, enabled=True, provider_id=None, aes_key="", channel="" -): - """Persist a meter driver entry and return its id.""" +def check_grpc_selection(details): + """Raise ProviderRegistrationError unless the contract supports being driven over gRPC. + + The spec fixes no gRPC port, so gRPC needs a grpc interface with a + target in x-meter-driver. Its ConfigureDriver message has fixed + fields, so the driver's /v1/requirements must list + heartbeat_period_duration and aes_key among the required fields. + """ + grpc_interface = next( + (interface for interface in details.get("interfaces") or [] if interface.get("type") == "grpc"), + None, + ) + if grpc_interface is None: + raise ProviderRegistrationError( + "gRPC is not available: the driver's x-meter-driver block advertises no grpc interface" + ) + if not (grpc_interface.get("target") or grpc_interface.get("address")): + raise ProviderRegistrationError( + "gRPC is not available: the driver's grpc interface advertises no target" + ) + required_names = { + field["name"] for field in details.get("driver_requirement_fields") or [] if field.get("required") + } + missing = [name for name in GRPC_INIT_REQUIRED_FIELDS if name not in required_names] + if missing: + raise ProviderRegistrationError( + "gRPC init (ConfigureDriver) needs the init fields {} but the driver's /v1/requirements " + "does not list: {}".format(", ".join(GRPC_INIT_REQUIRED_FIELDS), ", ".join(missing)) + ) + + +def save_provider_settings(service_url, selected_interface, enabled=True, provider_id=None): + """Persist a meter driver entry and return its id. + + A gRPC selection is refused (ProviderRegistrationError) unless the + contract advertises a grpc target and the init fields gRPC needs; any + other interface the contract does not advertise falls back to its + default interface. + """ details = validate_contract(service_url) selected_interface = (selected_interface or details["default_interface"]).strip().lower() valid_interfaces = {interface["type"] for interface in details.get("interfaces") or []} - if selected_interface not in valid_interfaces: + if selected_interface == "grpc": + check_grpc_selection(details) + elif selected_interface not in valid_interfaces: selected_interface = details["default_interface"] selected_interface_details = next( @@ -756,9 +1035,19 @@ def get_openapi_url(service_url): def _normalize_interface_metadata(base_url, spec): - """Extract the advertised interface inventory from a provider contract.""" - extension = spec.get("x-open-thunder") or {} - interface_entries = extension.get("interfaces") or [] + """Extract the advertised interface inventory from the contract's x-meter-driver block. + + Without the block, or without an http entry in it, one http interface + at the registered base URL is synthesized and becomes the default. + That leniency goes beyond the spec, which says the block lists the + active interfaces. + """ + extension = spec.get(DISCOVERY_EXTENSION) + if not isinstance(extension, dict): + extension = {} + interface_entries = extension.get("interfaces") + if not isinstance(interface_entries, list): + interface_entries = [] interfaces = [] seen_types = set() @@ -862,8 +1151,37 @@ def _fallback_interface_metadata(base_url, selected_interface=None): ) -def validate_contract(service_url, timeout=10.0): - """Fetch and validate the provider's OpenAPI contract.""" +def _normalize_path_template(path): + """Return a path with every {param} replaced by {} so parameter names do not matter.""" + return _PATH_PARAM_RE.sub("{}", path) + + +def _missing_required_operations(paths): + """Return "METHOD /path" labels for the spec's required operations a document lacks. + + A path matches regardless of its parameter names (/v1/nodes/{id} is + /v1/nodes/{node_id}); a trailing slash does not match. + """ + by_template = {} + for path, item in paths.items(): + if isinstance(path, str) and isinstance(item, dict): + by_template.setdefault(_normalize_path_template(path), item) + + missing = [] + for method, path in REQUIRED_CONTRACT_OPERATIONS: + item = by_template.get(_normalize_path_template(path), {}) + if not isinstance(item.get(method), dict): + missing.append("{} {}".format(method.upper(), path)) + return missing + + +def _fetch_contract_document(service_url, timeout=10.0): + """GET /openapi.json and check it is a document a compliant driver would serve. + + Returns `(base_url, openapi_url, document)`. The document must be a + JSON object with `info.title`, a `paths` object listing every required + operation, and, when present, an object-valued x-meter-driver block. + """ base_url = normalize_base_url(service_url) openapi_url = get_openapi_url(service_url) try: @@ -876,95 +1194,157 @@ def validate_contract(service_url, timeout=10.0): spec = response.json() except ValueError as exc: raise ProviderRegistrationError("driver returned invalid JSON") from exc + if not isinstance(spec, dict): + raise ProviderRegistrationError("driver OpenAPI document must be a JSON object") + + info = spec.get("info") + if not isinstance(info, dict) or not info.get("title"): + raise ProviderRegistrationError("driver contract missing info.title") - info = spec.get("info") or {} - paths = spec.get("paths") or {} - required_paths = ("/v1/commands", "/v1/events") - missing_paths = [path for path in required_paths if path not in paths] - if missing_paths: + paths = spec.get("paths") + if not isinstance(paths, dict): + raise ProviderRegistrationError("driver contract missing paths") + missing = _missing_required_operations(paths) + if missing: raise ProviderRegistrationError( - "driver contract missing required paths: {}".format(", ".join(missing_paths)) + "driver contract missing required routes: {}".format(", ".join(missing)) ) - name = info.get("title") - if not name: - raise ProviderRegistrationError("driver contract missing info.title") + extension = spec.get(DISCOVERY_EXTENSION) + if extension is not None and not isinstance(extension, dict): + raise ProviderRegistrationError("driver contract {} must be an object".format(DISCOVERY_EXTENSION)) - driver_requirement_fields = _extract_driver_requirement_fields(base_url, spec, timeout=timeout) + return base_url, openapi_url, spec + +def _contract_details(base_url, openapi_url, spec): + """Build the details dict for a fetched contract, without init-field discovery.""" + info = spec["info"] return { - "name": str(name), + "name": str(info["title"]), "base_url": base_url, "openapi_url": openapi_url, "service_version": str(info.get("version") or ""), - "driver_requirement_fields": driver_requirement_fields, - "driver_requirement_field_map": {field["name"]: field for field in driver_requirement_fields}, + "driver_requirement_fields": [], + "driver_requirement_field_map": {}, "vendor_option_fields": _extract_vendor_option_fields(spec), "vendor_option_field_map": _vendor_option_field_map(spec), **_normalize_interface_metadata(base_url, spec), } -def get_live_interface_details(service_url, selected_interface=None, timeout=2.0): - """Fetch the current interface inventory advertised by the provider.""" +def inspect_contract(service_url, timeout=10.0): + """Fetch and validate the driver's OpenAPI contract and discover its interfaces. + + One round trip (GET /openapi.json). `driver_requirement_fields` is + empty here: init-field discovery is a second round trip that + registration makes through `validate_contract`. + """ + base_url, openapi_url, spec = _fetch_contract_document(service_url, timeout=timeout) + return _contract_details(base_url, openapi_url, spec) + + +def validate_contract(service_url, timeout=10.0): + """Fetch and validate the driver's OpenAPI contract, then discover its init fields. + + Follows the spec's integration sequence (section 2): GET /openapi.json, + check it lists every required operation, read x-meter-driver, then GET + /v1/requirements typed by the document's InitRequest schema. A driver + missing any of that is not registrable. + """ + base_url, openapi_url, spec = _fetch_contract_document(service_url, timeout=timeout) + details = _contract_details(base_url, openapi_url, spec) + driver_requirement_fields = _extract_driver_requirement_fields(base_url, spec, timeout=timeout) + details["driver_requirement_fields"] = driver_requirement_fields + details["driver_requirement_field_map"] = {field["name"]: field for field in driver_requirement_fields} + return details + + +def _stored_driver_fields(provider): + """Return the field specs recorded in a saved provider's config file, if any.""" + if not provider: + return [] + try: + return list(_stored_field_specs(load_provider_runtime_settings(provider)).values()) + except DriverConfigError: + return [] + + +def get_live_interface_details(service_url, selected_interface=None, timeout=2.0, provider=None): + """Fetch the interface inventory the driver advertises right now. + + Only GET /openapi.json is called, so an advertised gRPC target is + never lost to a slow or failing /v1/requirements. When `provider` is + given, `driver_requirement_fields` come from the fields recorded in + its config file at registration. + """ base_url = normalize_base_url(service_url) try: - provider_data = validate_contract(base_url, timeout=timeout) + details = inspect_contract(base_url, timeout=timeout) except ProviderRegistrationError as exc: - details = _fallback_interface_metadata( - base_url, - selected_interface=selected_interface, - ) + details = _fallback_interface_metadata(base_url, selected_interface=selected_interface) details["error"] = str(exc) - return details - return _apply_selected_interface( - provider_data, - selected_interface=selected_interface, - ) + fields = _stored_driver_fields(provider) + details["driver_requirement_fields"] = fields + details["driver_requirement_field_map"] = {field["name"]: field for field in fields} + return _apply_selected_interface(details, selected_interface=selected_interface) def get_runtime_status(service_url, timeout=2.0, include_gateway_status=True): - """Check whether the provider service is currently reachable.""" + """Check driver liveness on GET /v1/healthz and, optionally, gateway state on GET /v1/status. + + Online means /v1/healthz answered 200 with the spec's HealthResponse + `{"ok": true}`. With `include_gateway_status`, the `connected` and + `gateway_type` of the JSON object /v1/status answers are reported; a + transport failure, invalid JSON or a body that is not an object there + leaves the driver online with no gateway. + """ base_url = normalize_base_url(service_url) healthz_url = base_url.rstrip("/") + "/v1/healthz" - legacy_health_url = base_url.rstrip("/") + "/health" status_url = base_url.rstrip("/") + "/v1/status" - urls = (healthz_url, legacy_health_url) - last_error = None - for url in urls: - try: - response = httpx.get(url, timeout=timeout) - response.raise_for_status() - status = { - "online": True, - "message": "online", - "checked_url": url, - "gateway_checked": bool(include_gateway_status), - } - if not include_gateway_status: - status["gateway_active"] = False - status["gateway_type"] = None - return status - try: - gateway_response = httpx.get(status_url, timeout=timeout) - gateway_response.raise_for_status() - gateway_data = gateway_response.json() - status["gateway_active"] = bool(gateway_data.get("connected")) - status["gateway_type"] = gateway_data.get("gateway_type") - except (httpx.HTTPError, ValueError): - status["gateway_checked"] = True - status["gateway_active"] = False - status["gateway_type"] = None - return status - except httpx.HTTPError as exc: - last_error = exc - return { - "online": False, - "message": str(last_error) if last_error is not None else "unreachable", + def offline(message): + return { + "online": False, + "message": message, + "checked_url": healthz_url, + "gateway_checked": bool(include_gateway_status), + "gateway_active": False, + "gateway_type": None, + } + + try: + response = httpx.get(healthz_url, timeout=timeout) + response.raise_for_status() + health = response.json() + except httpx.HTTPError as exc: + return offline(str(exc) or "unreachable") + except ValueError: + return offline("driver /v1/healthz response is not valid JSON") + if not isinstance(health, dict) or health.get("ok") is not True: + return offline('driver /v1/healthz did not answer {"ok": true}') + + status = { + "online": True, + "message": "online", "checked_url": healthz_url, "gateway_checked": bool(include_gateway_status), - "gateway_active": False, - "gateway_type": None, } + if not include_gateway_status: + status["gateway_active"] = False + status["gateway_type"] = None + return status + try: + gateway_response = httpx.get(status_url, timeout=timeout) + gateway_response.raise_for_status() + gateway_data = gateway_response.json() + except (httpx.HTTPError, ValueError): + gateway_data = None + if not isinstance(gateway_data, dict): + status["gateway_active"] = False + status["gateway_type"] = None + return status + status["gateway_active"] = bool(gateway_data.get("connected")) + status["gateway_type"] = gateway_data.get("gateway_type") + return status diff --git a/sparkmeter/config/providerform.py b/sparkmeter/config/providerform.py index 55dedad..d624805 100644 --- a/sparkmeter/config/providerform.py +++ b/sparkmeter/config/providerform.py @@ -43,38 +43,25 @@ def __init__(self, *args, **kwargs): selected_interface=( self.provider["selected_interface"] if self.provider is not None else None ), + provider=self.provider, ) if self.provider_details is not None: self._set_interface_choices(self.provider_details) - self._apply_vendor_option_labels() + self._apply_driver_field_labels() if not self.selected_interface.data: self.selected_interface.data = ( self.provider["selected_interface"] if self.provider is not None else None ) or self._default_selected_interface() - def vendor_option_spec(self, name): - """Return the normalized driver-requirement spec for a known field.""" + def driver_field_spec(self, name): + """Return the normalized spec of one field the driver asks for, if it does.""" provider_data = self.provider_details or {} return (provider_data.get("driver_requirement_field_map") or {}).get(name) - def supports_vendor_option(self, name): - """Whether the validated provider advertises the given vendor option.""" - return self.vendor_option_spec(name) is not None - - def vendor_option_description(self, name): - """Return help text for a vendor option field.""" - spec = self.vendor_option_spec(name) or {} - return spec.get("description") or "" - - def vendor_option_required(self, name): - """Whether the vendor option is required by the contract.""" - spec = self.vendor_option_spec(name) or {} - return bool(spec.get("required")) - - def vendor_option_fields(self): - """Return the contract-advertised driver requirement field list.""" + def driver_fields(self): + """Return the fields the driver asks for: /v1/requirements first, optional extras after.""" provider_data = self.provider_details or {} return provider_data.get("driver_requirement_fields") or [] @@ -114,13 +101,13 @@ def _default_selected_interface(self): return self.selected_interface.choices[0][0] return "http" - def _apply_vendor_option_labels(self): - """Update known vendor-option labels from the validated contract.""" - aes_key_spec = self.vendor_option_spec("aes_key") + def _apply_driver_field_labels(self): + """Update the known field labels from the validated contract.""" + aes_key_spec = self.driver_field_spec("aes_key") if aes_key_spec and aes_key_spec.get("label"): self.aes_key.label.text = aes_key_spec["label"] - channel_spec = self.vendor_option_spec("channel") + channel_spec = self.driver_field_spec("channel") if channel_spec and channel_spec.get("label"): self.channel.label.text = channel_spec["label"] @@ -133,7 +120,7 @@ def validate_service_url(self, field): try: self.provider_details = provider_settings.validate_contract(field.data) self._set_interface_choices(self.provider_details) - self._apply_vendor_option_labels() + self._apply_driver_field_labels() except provider_settings.ProviderRegistrationError as exc: raise ValidationError(str(exc)) @@ -149,6 +136,12 @@ def validate_selected_interface(self, field): if field.data not in valid_interfaces: raise ValidationError(_("Selected interface is not available from this driver.")) + if field.data == "grpc": + try: + provider_settings.check_grpc_selection(provider_data) + except provider_settings.ProviderRegistrationError as exc: + raise ValidationError(str(exc)) + def validate_aes_key(self, field): field.data = "" @@ -208,8 +201,8 @@ def __init__(self, *args, **kwargs): if not self.config_text.data: self.config_text.data = provider_settings.load_provider_config_text(self.provider) - def required_fields(self): - """Return the driver-required field list.""" + def driver_fields(self): + """Return the fields the driver asks for: /v1/requirements first, optional extras after.""" return (self.provider_details or {}).get("driver_requirement_fields") or [] def config_file_path(self): diff --git a/sparkmeter/config/templates/config-meter-driver-config-editor.html b/sparkmeter/config/templates/config-meter-driver-config-editor.html index dd6d397..9b80c70 100644 --- a/sparkmeter/config/templates/config-meter-driver-config-editor.html +++ b/sparkmeter/config/templates/config-meter-driver-config-editor.html @@ -23,12 +23,17 @@

{{ _('Driver') }}: {{ provider_details.name if provider_details and provider_details.name else provider.name }}

{{ _('File') }}: {{ form.config_file_path() }}

- {% if form.required_fields() %} -

{{ _('Required fields') }}:

+ {% if form.driver_fields() %} +

{{ _('Driver fields') }}: