diff --git a/CHANGELOG.md b/CHANGELOG.md index 9e5c6889..936e2679 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,12 +20,19 @@ All notable changes to this project will be documented in this file. - The RBAC ServiceAccount and RoleBinding are now built with the operator-rs `v2::rbac` functions and carry the full set of recommended labels ([#861]). - All product containers now run with `securityContext.runAsNonRoot` set to `true` to improve security ([#871]). +- The user-info-fetcher Keycloak and Entra backends now cache their OAuth2 access token for the + lifetime the identity provider reports, instead of minting a new one for every user lookup. This + removes one round trip per lookup. If the provider rejects the cached token before it expires, it is + re-minted and the lookup is retried once ([#863]). ### Fixed - Fix a longstanding problem of including empty `categories`, `shortNames` and `additionalPrinterColumns` in the CRDs, which could cause problems with GitOps tools (e.g. ArgoCD) reporting a diff in the custom resources. See [our internal issue](https://github.com/stackabletech/hdfs-operator/issues/626) and [the fix](https://github.com/kube-rs/kube/pull/2042) for details ([#871]). +- The file logs of the user-info-fetcher and resource-info-fetcher sidecars are now collected by the + Vector agent. Both sidecars log below `/stackable/log`, but did not mount the shared `log` volume, + so their logs were unreachable for Vector and not accounted for in the volume's size limit ([#863]). [#852]: https://github.com/stackabletech/opa-operator/pull/852 [#861]: https://github.com/stackabletech/opa-operator/pull/861 diff --git a/Cargo.lock b/Cargo.lock index 07d200d7..b0851e10 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1699,6 +1699,7 @@ name = "info-fetcher-commons" version = "0.0.0-dev" dependencies = [ "axum", + "futures", "hyper", "native-tls", "reqwest", @@ -3740,6 +3741,7 @@ dependencies = [ "tokio", "tracing", "url", + "wiremock", ] [[package]] diff --git a/Cargo.nix b/Cargo.nix index 10460b02..4b3ad0cb 100644 --- a/Cargo.nix +++ b/Cargo.nix @@ -5497,6 +5497,12 @@ rec { packageId = "tracing"; } ]; + devDependencies = [ + { + name = "futures"; + packageId = "futures"; + } + ]; }; "ipnet" = rec { @@ -12385,6 +12391,10 @@ rec { name = "rstest"; packageId = "rstest"; } + { + name = "wiremock"; + packageId = "wiremock"; + } ]; }; diff --git a/Cargo.toml b/Cargo.toml index 78011986..b64a7e7d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -49,7 +49,6 @@ tar = "0.4" tokio = { version = "1.53", features = ["full"] } tracing = "0.1" url = "2.5" -urlencoding = "2.1" uuid = "1.24" wiremock = "0.6" diff --git a/docs/modules/opa/pages/usage-guide/resource-info-fetcher.adoc b/docs/modules/opa/pages/usage-guide/resource-info-fetcher.adoc index a75f5526..f12999f4 100644 --- a/docs/modules/opa/pages/usage-guide/resource-info-fetcher.adoc +++ b/docs/modules/opa/pages/usage-guide/resource-info-fetcher.adoc @@ -92,11 +92,22 @@ The naming is intentionally product-agnostic, so that one function serves the eq `rawIdentifierResourceInfo` is the escape hatch for resources the functions above do not cover: it passes the identifier to the backend as-is. For DataHub that is a URN, such as `urn:li:chart:(superset,my-namespace/my-superset.1)`. +Note that metadata is only read from the entity types the functions above map to: datasets, containers, charts and dashboards. +Any other type (a `dataJob` or one of the ML entities, say) comes back empty, which looks just like a resource that has no metadata. +The resource-info-fetcher logs a warning naming the entity type when this happens. + The first two arguments are the same everywhere: * `system` is the kind of product the resource lives in, for example `trino`, `kafka` or `superset`. DataHub calls this the _data platform_. * `instance` identifies _which_ deployment of that product, for example `my-namespace/my-trino`. DataHub calls this the _platform instance_, and the value must match the `platform_instance` of the ingestion source that produced the metadata. +The `id` taken by `dashboardResourceInfo` and `chartResourceInfo` is whatever the product identifies the resource by, and is passed through as an opaque string. +Superset numbers its dashboards and charts, but products that name them instead work just as well. + +Arguments are limited to 1024 bytes, and may not contain `,`, `(` or `)`. +DataHub delimits the parts of a URN with those characters, so a resource whose name contains one cannot be addressed at all. +Either way the lookup is rejected with `400 Bad Request` instead of being sent to DataHub. `rawIdentifierResourceInfo` is exempt, as a URN necessarily contains them. + The DataHub environment (fabric) is deliberately *not* an argument: it describes how the catalog was populated rather than the resource being authorized, so it is configured once on the OpaCluster (see `env` above) instead of being passed in by every Rego rule. An example of the returned structure: @@ -129,6 +140,13 @@ An example of the returned structure: } ---- +[NOTE] +==== +DataHub models data product membership as graph edges rather than a field on the asset, so `dataProducts` is fetched as a single page of up to 1000 entries. +An asset is expected to belong to one or two, so this should not be reachable in practice. +If it ever is, the lookup fails with an error rather than returning the first 1000: a policy evaluating data product membership has no way to tell a truncated list from a complete one, and would silently decide on partial metadata. +==== + === Debug request To debug the resource-info-fetcher you can `curl` its API for a given resource. @@ -174,6 +192,68 @@ allow if { } ---- -A resource the backend does not know about is not reported as an error: the resource-info-fetcher returns a record with empty `tags`, `owners` and `dataProducts` and a `null` `domain`. -Prefer rules that require a positive signal, like the one above, which denies access in that case. -A rule that merely excludes a tag would instead grant access to every resource missing from the backend. +=== Metadata is not inherited from parent containers + +Tags, domains and data products are read from the addressed resource only. +Tagging the schema `tpch.sf1` as `pii` does not make `tableResourceInfo` report `pii` for the tables inside it. + +This is deliberate: whether a tag applies to a container's children is a property of your policy, not of the resource. +`pii` plausibly cascades, `deprecated` or an owning team plausibly do not, and we cannot tell which is which. +Merging them would also leave a rule unable to ask whether _this_ table is tagged. + +Express the inheritance you want in the rule instead, using one function per level: + +[source,rego] +---- +package test + +import data.stackable.opa.resourceinfo.v1 as resourceinfo + +default allow := false + +# The table itself is marked public, ... +allow if { + table := resourceinfo.tableResourceInfo("trino", "my-namespace/my-trino", input.catalog, input.schema, input.table) + some tag in table.tags + tag.urn == "urn:li:tag:public" +} + +# ... or the schema containing it is, which this rule chooses to extend to its tables. +allow if { + schema := resourceinfo.schemaResourceInfo("trino", "my-namespace/my-trino", input.catalog, input.schema) + some tag in schema.tags + tag.urn == "urn:li:tag:public" +} +---- + +Each lookup is cached and served over the loopback interface, so consulting an extra level costs little. + +=== Behaviour when metadata is unavailable + +[WARNING] +==== +A failed metadata lookup does not deny access by itself. +Only the shape of your rule decides that. +==== + +A lookup comes back without the metadata a rule expects in two cases: + +Unknown resource:: +The resource was never ingested into the catalog. +This is not an error, so the answer is `200 OK` with empty `tags`, `owners` and `dataProducts`, and a `null` `domain`. + +Backend unavailable:: +DataHub is down or unreachable, or the Personal Access Token expired or was revoked. +The answer is then an HTTP error status with `{"error": {"message": "...", "causes": ["..."]}}` instead of a metadata record. + +Either way the rule finds no `tags` to match on, so any expression reading them becomes undefined: + +* A rule keyed on a *positive* signal, like the `allow` example above, becomes undefined and therefore denies. This is what you want. +* A rule keyed on the *absence* of a signal (`deny` if tagged `pii`, everything else allowed) also becomes undefined, and an undefined `deny` means *not denied*. A DataHub outage then grants access to every resource, `pii` included. + +So always require a positive signal. +The resource-info-fetcher cannot know whether an empty record should mean allow or deny for your policy, so it does not paper over the difference. + +A failure is cached for a few seconds, well below `entryTimeToLive`, so a lookup that keeps failing neither queries the backend nor logs on every request. +An attempt that reaches the backend and fails is logged at `WARN`, so an unavailable backend shows up in the logs (see xref:opa:usage-guide/logging.adoc[]). +A lookup rejected because of the request itself, such as an identifier no URN can express, is logged at `DEBUG` instead: it says nothing about the health of the backend, and any user who can name a resource can produce those at will. diff --git a/rust/info-fetcher-commons/Cargo.toml b/rust/info-fetcher-commons/Cargo.toml index aa246ee2..90407b7d 100644 --- a/rust/info-fetcher-commons/Cargo.toml +++ b/rust/info-fetcher-commons/Cargo.toml @@ -21,3 +21,6 @@ serde_json.workspace = true snafu.workspace = true tracing.workspace = true tokio.workspace = true + +[dev-dependencies] +futures.workspace = true diff --git a/rust/info-fetcher-commons/src/utils/http.rs b/rust/info-fetcher-commons/src/utils/http.rs index 31ea5f2b..4557da1e 100644 --- a/rust/info-fetcher-commons/src/utils/http.rs +++ b/rust/info-fetcher-commons/src/utils/http.rs @@ -59,6 +59,28 @@ pub async fn send_json_request(req: RequestBuilder) -> Resu serde_json::from_str(&json).context(ParseJsonSnafu) } +/// Whether `error`, or any error it wraps, is a `401 Unauthorized` answer from a backend. +/// +/// Walks the source chain because backends wrap [`Error`] in their own error types, so the 401 is +/// never the outermost error by the time a caller gets to decide whether to re-authenticate. +pub fn is_unauthorized(error: &(dyn std::error::Error + 'static)) -> bool { + std::iter::successors(Some(error), |error| error.source()) + .filter_map(|error| error.downcast_ref::()) + .any(|error| error.status() == Some(StatusCode::UNAUTHORIZED)) +} + +impl Error { + /// The status code the backend answered with, or [`None`] if the failure happened before there was + /// a response to read a status off. + pub fn status(&self) -> Option { + match self { + Self::HttpErrorResponse { status, .. } => Some(*status), + Self::HttpErrorResponseUndecodableText { status, .. } => Some(*status), + Self::HttpRequest { .. } | Self::ParseJson { .. } => None, + } + } +} + /// Wraps a Response into a Result. If there is an HTTP Client or Server error, /// extract the HTTP body (if possible) to be used as context in the returned Err. /// This is done this because the `Response::error_for_status()` method Err variant @@ -84,3 +106,62 @@ async fn error_for_status(response: Response) -> Result { } Ok(response) } + +#[cfg(test)] +mod tests { + use snafu::IntoError; + + use super::*; + + /// Backends wrap transport errors in their own error types, sometimes several layers deep, so the + /// check has to walk the source chain instead of inspecting the outermost error. + #[derive(Snafu, Debug)] + #[snafu(display("failed to fetch the user"))] + struct FetchUser { + source: Error, + } + + #[derive(Snafu, Debug)] + #[snafu(display("failed to get user info"))] + struct GetUserInfo { + source: FetchUser, + } + + /// A backend response with `status`, wrapped the way a backend would wrap it. + fn wrapped_response(status: StatusCode) -> GetUserInfo { + let response = Error::HttpErrorResponse { + status, + url: "https://keycloak.example.com/admin/realms/my-realm/users/".to_owned(), + text: "denied".to_owned(), + }; + + GetUserInfoSnafu.into_error(FetchUserSnafu.into_error(response)) + } + + #[test] + fn a_wrapped_unauthorized_response_is_detected() { + assert!(is_unauthorized(&wrapped_response(StatusCode::UNAUTHORIZED))); + } + + /// Only a 401 means "your token is no good". A 403 says the token was understood and the actor is + /// not allowed, which re-minting cannot fix. + #[test] + fn other_error_responses_are_not_unauthorized() { + assert!(!is_unauthorized(&wrapped_response(StatusCode::FORBIDDEN))); + assert!(!is_unauthorized(&wrapped_response( + StatusCode::INTERNAL_SERVER_ERROR + ))); + } + + /// A request that never got an answer has no status to look at. + #[test] + fn errors_without_a_response_are_not_unauthorized() { + let error = Error::ParseJson { + source: serde_json::from_str::("not json") + .expect_err("the input is not valid JSON"), + }; + + assert_eq!(error.status(), None); + assert!(!is_unauthorized(&error)); + } +} diff --git a/rust/info-fetcher-commons/src/utils/mod.rs b/rust/info-fetcher-commons/src/utils/mod.rs index cb1f965d..4ad25bd9 100644 --- a/rust/info-fetcher-commons/src/utils/mod.rs +++ b/rust/info-fetcher-commons/src/utils/mod.rs @@ -1,2 +1,3 @@ pub mod http; pub mod tls; +pub mod token; diff --git a/rust/info-fetcher-commons/src/utils/token.rs b/rust/info-fetcher-commons/src/utils/token.rs new file mode 100644 index 00000000..3735f326 --- /dev/null +++ b/rust/info-fetcher-commons/src/utils/token.rs @@ -0,0 +1,238 @@ +//! Caching of the bearer tokens the info-fetcher backends authenticate with. + +use std::{ + future::Future, + time::{Duration, Instant}, +}; + +use tokio::sync::RwLock; + +/// How long before its stated expiry a token stops being handed out. +/// +/// A token is minted, then travels to the backend and is validated there, so handing out one that is +/// about to expire risks it being rejected mid-request. Refreshing slightly early avoids that without +/// needing to know anything about the backend's clock. +pub const EXPIRY_MARGIN: Duration = Duration::from_secs(30); + +/// A freshly minted bearer token. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct MintedToken { + pub token: String, + + /// How long the token remains valid, as reported by whoever issued it (for OAuth: `expires_in`). + /// + /// [`None`] when the issuer does not say. The token is then used but not cached, because we have + /// no basis for deciding when it goes stale. + pub lifetime: Option, +} + +/// A bearer token that is minted on demand and kept until shortly before it expires. +/// +/// Backends previously minted a token for every single request, which doubled the round trips per +/// lookup and threw the issuer's `expires_in` away. This caches the token for the lifetime the issuer +/// reported, and lets the caller drop it early via [`CachedToken::invalidate`] when the backend +/// rejects it. A token can stop working before its stated expiry, for example by being revoked. +#[derive(Debug, Default)] +pub struct CachedToken { + cached: RwLock>, +} + +#[derive(Debug)] +struct Entry { + token: String, + + /// The stated expiry, already brought forward by [`EXPIRY_MARGIN`]. + usable_until: Instant, +} + +impl CachedToken { + pub fn new() -> Self { + Self::default() + } + + /// Returns a usable token, minting one with `mint` if the cache has none or the cached one is + /// within [`EXPIRY_MARGIN`] of expiring. + /// + /// Concurrent callers that all miss the cache do not each mint: the first one to get the write + /// lock mints while the others wait, and they then find its result in the cache. Without that, a + /// burst of requests arriving just after a token expired would produce a burst of token requests. + pub async fn get(&self, mint: F) -> Result + where + F: FnOnce() -> Fut, + Fut: Future>, + { + if let Some(token) = Self::usable_token(&*self.cached.read().await) { + return Ok(token); + } + + // The read lock is released above, so another caller may have minted in between. Hence the + // second look before minting ourselves. + let mut cached = self.cached.write().await; + if let Some(token) = Self::usable_token(&cached) { + return Ok(token); + } + + let MintedToken { token, lifetime } = mint().await?; + + // Only cache a token we know is still usable for a worthwhile amount of time. A failed mint + // returns above, so nothing is cached in that case either. + *cached = lifetime + .and_then(|lifetime| lifetime.checked_sub(EXPIRY_MARGIN)) + .map(|usable_for| Entry { + token: token.clone(), + usable_until: Instant::now() + usable_for, + }); + + Ok(token) + } + + /// Drops the cached token, so the next [`CachedToken::get`] mints a fresh one. + /// + /// Call this when the backend rejects the token, which is the only way to find out that it stopped + /// being valid ahead of its stated expiry. + pub async fn invalidate(&self) { + *self.cached.write().await = None; + } + + /// The cached token, if there is one and it is not within [`EXPIRY_MARGIN`] of expiring. + fn usable_token(cached: &Option) -> Option { + let entry = cached.as_ref()?; + + (entry.usable_until > Instant::now()).then(|| entry.token.clone()) + } +} + +#[cfg(test)] +mod tests { + use std::{ + sync::atomic::{AtomicUsize, Ordering}, + time::Duration, + }; + + use super::*; + + /// Counts how often a token was minted, so the tests can assert on cache hits rather than on + /// timing. + struct Minter { + calls: AtomicUsize, + lifetime: Option, + } + + impl Minter { + fn new(lifetime: Option) -> Self { + Self { + calls: AtomicUsize::new(0), + lifetime, + } + } + + async fn mint(&self) -> Result { + let call = self.calls.fetch_add(1, Ordering::SeqCst); + + Ok(MintedToken { + token: format!("token-{call}"), + lifetime: self.lifetime, + }) + } + + fn calls(&self) -> usize { + self.calls.load(Ordering::SeqCst) + } + } + + /// A lifetime comfortably longer than [`EXPIRY_MARGIN`], so the token stays usable for the whole + /// test without any waiting. + fn long_lifetime() -> Option { + Some(EXPIRY_MARGIN + Duration::from_secs(600)) + } + + #[tokio::test] + async fn token_is_minted_once_and_then_served_from_the_cache() { + let minter = Minter::new(long_lifetime()); + let cached = CachedToken::new(); + + let first = cached.get(|| minter.mint()).await.expect("minting works"); + let second = cached.get(|| minter.mint()).await.expect("minting works"); + + assert_eq!(first, "token-0"); + assert_eq!(second, "token-0"); + assert_eq!(minter.calls(), 1); + } + + #[tokio::test] + async fn invalidating_forces_the_next_get_to_mint() { + let minter = Minter::new(long_lifetime()); + let cached = CachedToken::new(); + + let first = cached.get(|| minter.mint()).await.expect("minting works"); + cached.invalidate().await; + let second = cached.get(|| minter.mint()).await.expect("minting works"); + + assert_eq!(first, "token-0"); + assert_eq!(second, "token-1"); + assert_eq!(minter.calls(), 2); + } + + /// Without a stated lifetime we have no idea how long the token is good for, so we must not keep + /// it. That degrades to minting per request, which is what the backends did before caching. + #[tokio::test] + async fn a_token_without_a_lifetime_is_not_cached() { + let minter = Minter::new(None); + let cached = CachedToken::new(); + + cached.get(|| minter.mint()).await.expect("minting works"); + cached.get(|| minter.mint()).await.expect("minting works"); + + assert_eq!(minter.calls(), 2); + } + + /// A token that expires within the safety margin has no usable lifetime left, so caching it would + /// only hand out a token that is about to be rejected. + #[tokio::test] + async fn a_token_expiring_within_the_margin_is_not_cached() { + let minter = Minter::new(Some(EXPIRY_MARGIN)); + let cached = CachedToken::new(); + + cached.get(|| minter.mint()).await.expect("minting works"); + cached.get(|| minter.mint()).await.expect("minting works"); + + assert_eq!(minter.calls(), 2); + } + + /// Concurrent lookups that all miss the cache must not each mint a token: the backend would see a + /// burst of token requests every time the cached one expires. + #[tokio::test] + async fn concurrent_gets_mint_only_once() { + let minter = Minter::new(long_lifetime()); + let cached = CachedToken::new(); + + let tokens = futures::future::join_all((0..20).map(|_| { + cached.get(|| async { + // Yield inside the critical section, so the tasks actually overlap. + tokio::time::sleep(Duration::from_millis(20)).await; + minter.mint().await + }) + })) + .await; + + for token in tokens { + assert_eq!(token.expect("minting works"), "token-0"); + } + assert_eq!(minter.calls(), 1); + } + + /// A failed mint must not be cached, and must not poison a later attempt. + #[tokio::test] + async fn a_failed_mint_is_not_cached() { + let minter = Minter::new(long_lifetime()); + let cached = CachedToken::new(); + + let failed = cached + .get(|| async { Err::("no token for you".to_owned()) }) + .await; + let succeeded = cached.get(|| minter.mint()).await; + + assert_eq!(failed, Err("no token for you".to_owned())); + assert_eq!(succeeded, Ok("token-0".to_owned())); + } +} diff --git a/rust/operator-binary/src/controller/build/properties/product_logging/vector.yaml b/rust/operator-binary/src/controller/build/properties/product_logging/vector.yaml index eb01ff24..244b4f6d 100644 --- a/rust/operator-binary/src/controller/build/properties/product_logging/vector.yaml +++ b/rust/operator-binary/src/controller/build/properties/product_logging/vector.yaml @@ -20,7 +20,7 @@ sources: include: - ${LOG_DIR}/*/*.stderr.log - # Logs of the Stackable Rust sidecars (bundle-builder, user-info-fetcher) + # Logs of the Stackable Rust sidecars (bundle-builder, user-info-fetcher, resource-info-fetcher) files_tracing_rs: type: file include: diff --git a/rust/operator-binary/src/controller/build/resource/daemonset/mod.rs b/rust/operator-binary/src/controller/build/resource/daemonset/mod.rs index e6ab8c44..40226952 100644 --- a/rust/operator-binary/src/controller/build/resource/daemonset/mod.rs +++ b/rust/operator-binary/src/controller/build/resource/daemonset/mod.rs @@ -28,7 +28,9 @@ use stackable_operator::{ ResourceRequirements, }, }, - apimachinery::pkg::{apis::meta::v1::LabelSelector, util::intstr::IntOrString}, + apimachinery::pkg::{ + api::resource::Quantity, apis::meta::v1::LabelSelector, util::intstr::IntOrString, + }, }, memory::{BinaryMultiple, MemoryQuantity}, product_logging::{ @@ -96,6 +98,8 @@ const LIVENESS_PROBE_INITIAL_DELAY_SECONDS: i32 = 30; const CONSOLE_LOG_LEVEL_ENV: &str = "CONSOLE_LOG_LEVEL"; const FILE_LOG_LEVEL_ENV: &str = "FILE_LOG_LEVEL"; const FILE_LOG_DIRECTORY_ENV: &str = "FILE_LOG_DIRECTORY"; +const FILE_LOG_ROTATION_PERIOD_ENV: &str = "FILE_LOG_ROTATION_PERIOD"; +const FILE_LOG_MAX_FILES_ENV: &str = "FILE_LOG_MAX_FILES"; const KUBERNETES_NODE_NAME_ENV: &str = "KUBERNETES_NODE_NAME"; const KUBERNETES_CLUSTER_DOMAIN_ENV: &str = "KUBERNETES_CLUSTER_DOMAIN"; @@ -133,6 +137,31 @@ const MAX_PREPARE_LOG_FILE_SIZE: MemoryQuantity = MemoryQuantity { unit: BinaryMultiple::Mebi, }; +// The info-fetcher sidecars log one line per startup and one per failed request, so they are far +// less chatty than the bundle-builder. They are only budgeted for when the corresponding sidecar is +// actually part of the Pod, see `log_volume_size_limit`. +const MAX_INFO_FETCHER_LOG_FILE_SIZE: MemoryQuantity = MemoryQuantity { + value: 5.0, + unit: BinaryMultiple::Mebi, +}; + +// Rotation of the file logs written by the Stackable Rust containers. Nothing else bounds those +// files: the Vector agent only reads them, and they are written into the shared `log` volume, whose +// `sizeLimit` evicts the whole Pod (OPA included) once it is exceeded. +// +// This bounds how many files are kept, not how large they get, so the budgeted MAX_*_LOG_FILE_SIZE +// above stays an estimate. Products get a size-based rotating appender from operator-rs, but +// stackable-telemetry has no size-based policy, see +// https://github.com/stackabletech/opa-operator/issues/606. +// +// Hence the short period: the worst case is a sustained backend outage, where the info-fetchers log +// one line per failed request, and the retained volume is the log rate times the period times the +// file count. Keeping minutes rather than hours is what makes that survivable. Little is lost by it, +// because the Vector agent ships the lines as they are written, and the console logs (captured by the +// container runtime) are unaffected either way. +const FILE_LOG_ROTATION_PERIOD: &str = "minutely"; +const FILE_LOG_MAX_FILES: u32 = 5; + #[derive(Snafu, Debug)] pub enum Error { #[snafu(display("failed to configure graceful shutdown"))] @@ -385,13 +414,7 @@ pub fn build_server_rolegroup_daemonset( VolumeBuilder::new(LOG_VOLUME_NAME.as_ref()) .empty_dir(EmptyDirVolumeSource { medium: None, - size_limit: Some(product_logging::framework::calculate_log_volume_size_limit( - &[ - MAX_OPA_BUNDLE_BUILDER_LOG_FILE_SIZE, - MAX_OPA_LOG_FILE_SIZE, - MAX_PREPARE_LOG_FILE_SIZE, - ], - )), + size_limit: Some(log_volume_size_limit(cluster)), }) .build(), ) @@ -507,9 +530,33 @@ pub fn build_server_rolegroup_daemonset( }) } +/// The size limit of the shared `log` [`EmptyDirVolumeSource`], which has to accommodate the log +/// files of every container that mounts it. The always-present containers are budgeted for +/// unconditionally; the optional info-fetcher sidecars only when the cluster configures them. +fn log_volume_size_limit(cluster: &ValidatedCluster) -> Quantity { + let mut max_log_file_sizes = vec![ + MAX_OPA_BUNDLE_BUILDER_LOG_FILE_SIZE, + MAX_OPA_LOG_FILE_SIZE, + MAX_PREPARE_LOG_FILE_SIZE, + ]; + + if cluster.cluster_config.user_info.is_some() { + max_log_file_sizes.push(MAX_INFO_FETCHER_LOG_FILE_SIZE); + } + if cluster.cluster_config.resource_info.is_some() { + max_log_file_sizes.push(MAX_INFO_FETCHER_LOG_FILE_SIZE); + } + + product_logging::framework::calculate_log_volume_size_limit(&max_log_file_sizes) +} + /// Env variables that are need to run stackable Rust binaries, such as /// * opa-bundle-builder /// * user-info-fetcher +/// * resource-info-fetcher +/// +/// Note that [`FILE_LOG_DIRECTORY_ENV`] points below [`STACKABLE_LOG_DIR`], so every container this +/// is applied to has to mount the `log` volume for the Vector agent to see its logs. fn add_stackable_rust_cli_env_vars( container_builder: &mut ContainerBuilder, cluster_info: &KubernetesClusterInfo, @@ -524,6 +571,8 @@ fn add_stackable_rust_cli_env_vars( FILE_LOG_DIRECTORY_ENV, format!("{STACKABLE_LOG_DIR}/{container}",), ) + .add_env_var(FILE_LOG_ROTATION_PERIOD_ENV, FILE_LOG_ROTATION_PERIOD) + .add_env_var(FILE_LOG_MAX_FILES_ENV, FILE_LOG_MAX_FILES.to_string()) .add_env_var_from_source( KUBERNETES_NODE_NAME_ENV, EnvVarSource { @@ -1008,7 +1057,7 @@ mod tests { ); } - fn uif_container(ds: &DaemonSet) -> Container { + fn container_by_name(ds: &DaemonSet, name: &str) -> Container { ds.spec .as_ref() .unwrap() @@ -1018,11 +1067,15 @@ mod tests { .unwrap() .containers .iter() - .find(|c| c.name == "user-info-fetcher") - .expect("the user-info-fetcher container should exist") + .find(|c| c.name == name) + .unwrap_or_else(|| panic!("the {name} container should exist")) .clone() } + fn uif_container(ds: &DaemonSet) -> Container { + container_by_name(ds, "user-info-fetcher") + } + fn env_var(container: &Container, name: &str) -> String { container .env @@ -1140,4 +1193,117 @@ mod tests { "/stackable/credentials" ); } + + /// A cluster running both info-fetcher sidecars, so their shared wiring can be asserted in one go. + fn cluster_with_both_info_fetchers() -> ValidatedCluster { + validated_cluster_from_spec(json!({ + "image": { "productVersion": "1.2.3" }, + "clusterConfig": { + "userInfo": { + "backend": { + "experimentalXfscAas": { + "hostname": "aas.default.svc.cluster.local", + "port": 5000, + } + } + }, + "resourceInfo": { + "backend": { + "dataHub": { + "hostname": "datahub-gms.default.svc.cluster.local", + "credentialsSecretName": "datahub-credentials", + } + } + }, + }, + "servers": { "roleGroups": { "default": {} } }, + })) + } + + fn log_volume_size_limit(ds: &DaemonSet) -> Quantity { + ds.spec + .as_ref() + .unwrap() + .template + .spec + .as_ref() + .unwrap() + .volumes + .as_ref() + .unwrap() + .iter() + .find(|volume| volume.name == LOG_VOLUME_NAME.as_ref()) + .expect("the log volume should exist") + .empty_dir + .as_ref() + .expect("the log volume should be an emptyDir") + .size_limit + .clone() + .expect("the log volume should have a size limit") + } + + /// Both sidecars write their file logs below `STACKABLE_LOG_DIR`, so they have to mount the `log` + /// volume. Otherwise the logs land in the container's own filesystem, where the Vector agent + /// (which only reads the shared volume) cannot see them. + #[test] + fn info_fetcher_sidecars_mount_the_log_volume() { + let ds = build(&cluster_with_both_info_fetchers()); + + for container_name in ["user-info-fetcher", "resource-info-fetcher"] { + let container = container_by_name(&ds, container_name); + assert_eq!( + mount_path(&container, "log"), + "/stackable/log", + "{container_name} should mount the log volume" + ); + // The directory the sidecar logs into must be inside the mounted volume. + assert_eq!( + env_var(&container, "FILE_LOG_DIRECTORY"), + format!("/stackable/log/{container_name}") + ); + } + } + + /// Nothing else bounds these files. Vector only reads them, and the products' log frameworks + /// (which operator-rs configures with a size-based rotating appender) are not involved here, so + /// the writer has to be told to roll them over. + /// + /// The period is asserted because it is what caps the damage during a sustained backend outage, + /// when the info-fetchers log one line per failed request. + #[test] + fn the_rust_containers_rotate_their_file_logs() { + let ds = build(&cluster_with_both_info_fetchers()); + + for container in [ + "bundle-builder", + "user-info-fetcher", + "resource-info-fetcher", + ] { + let container = container_by_name(&ds, container); + assert_eq!(env_var(&container, "FILE_LOG_ROTATION_PERIOD"), "minutely"); + assert_eq!(env_var(&container, "FILE_LOG_MAX_FILES"), "5"); + } + } + + /// The sidecars share the `log` volume with the other containers, so their log files have to be + /// budgeted for in its size limit as well. + #[test] + fn log_volume_size_limit_accounts_for_the_info_fetcher_sidecars() { + let without_sidecars = build(&validated_cluster_from_spec(json!({ + "image": { "productVersion": "1.2.3" }, + "servers": { "roleGroups": { "default": {} } }, + }))); + let with_sidecars = build(&cluster_with_both_info_fetchers()); + + // prepare + opa + bundle-builder + assert_eq!( + log_volume_size_limit(&without_sidecars), + Quantity("108Mi".to_owned()) + ); + // ... plus the two info-fetcher sidecars + assert_eq!( + log_volume_size_limit(&with_sidecars), + Quantity("138Mi".to_owned()) + ); + } } diff --git a/rust/operator-binary/src/controller/build/resource/daemonset/resource_info_fetcher.rs b/rust/operator-binary/src/controller/build/resource/daemonset/resource_info_fetcher.rs index 21f64222..b4bc10bf 100644 --- a/rust/operator-binary/src/controller/build/resource/daemonset/resource_info_fetcher.rs +++ b/rust/operator-binary/src/controller/build/resource/daemonset/resource_info_fetcher.rs @@ -16,9 +16,10 @@ use crate::controller::{ build::{ self, resource::daemonset::{ - CONFIG_DIR, CONFIG_VOLUME_NAME, RESOURCE_INFO_FETCHER_CREDENTIALS_DIR, - RESOURCE_INFO_FETCHER_CREDENTIALS_VOLUME_NAME, add_stackable_rust_cli_env_vars, - container_name, sidecar_container_log_level, sidecar_resource_requirements, + CONFIG_DIR, CONFIG_VOLUME_NAME, LOG_VOLUME_NAME, RESOURCE_INFO_FETCHER_CREDENTIALS_DIR, + RESOURCE_INFO_FETCHER_CREDENTIALS_VOLUME_NAME, STACKABLE_LOG_DIR, + add_stackable_rust_cli_env_vars, container_name, sidecar_container_log_level, + sidecar_resource_requirements, }, }, }; @@ -66,6 +67,11 @@ pub fn add_resource_info_fetcher_sidecar( .add_env_var("CREDENTIALS_DIR", RESOURCE_INFO_FETCHER_CREDENTIALS_DIR) .add_volume_mount(CONFIG_VOLUME_NAME.as_ref(), CONFIG_DIR) .context(AddVolumeMountSnafu)? + // The sidecar writes its file logs below this directory (see + // `add_stackable_rust_cli_env_vars`). They have to land on the shared log volume, + // because that is the only place the Vector agent collects them from. + .add_volume_mount(LOG_VOLUME_NAME.as_ref(), STACKABLE_LOG_DIR) + .context(AddVolumeMountSnafu)? .resources(sidecar_resource_requirements()); add_stackable_rust_cli_env_vars( &mut cb_rif, diff --git a/rust/operator-binary/src/controller/build/resource/daemonset/user_info_fetcher.rs b/rust/operator-binary/src/controller/build/resource/daemonset/user_info_fetcher.rs index 31b4f542..81d59ecf 100644 --- a/rust/operator-binary/src/controller/build/resource/daemonset/user_info_fetcher.rs +++ b/rust/operator-binary/src/controller/build/resource/daemonset/user_info_fetcher.rs @@ -22,10 +22,11 @@ use crate::controller::{ build::{ self, resource::daemonset::{ - CONFIG_DIR, CONFIG_VOLUME_NAME, USER_INFO_FETCHER_CREDENTIALS_DIR, - USER_INFO_FETCHER_CREDENTIALS_VOLUME_NAME, USER_INFO_FETCHER_KERBEROS_DIR, - USER_INFO_FETCHER_KERBEROS_VOLUME_NAME, add_stackable_rust_cli_env_vars, - container_name, sidecar_container_log_level, sidecar_resource_requirements, + CONFIG_DIR, CONFIG_VOLUME_NAME, LOG_VOLUME_NAME, STACKABLE_LOG_DIR, + USER_INFO_FETCHER_CREDENTIALS_DIR, USER_INFO_FETCHER_CREDENTIALS_VOLUME_NAME, + USER_INFO_FETCHER_KERBEROS_DIR, USER_INFO_FETCHER_KERBEROS_VOLUME_NAME, + add_stackable_rust_cli_env_vars, container_name, sidecar_container_log_level, + sidecar_resource_requirements, }, }, }; @@ -93,6 +94,11 @@ pub fn add_user_info_fetcher_sidecar( .add_env_var("CREDENTIALS_DIR", USER_INFO_FETCHER_CREDENTIALS_DIR) .add_volume_mount(CONFIG_VOLUME_NAME.as_ref(), CONFIG_DIR) .context(AddVolumeMountSnafu)? + // The sidecar writes its file logs below this directory (see + // `add_stackable_rust_cli_env_vars`). They have to land on the shared log volume, + // because that is the only place the Vector agent collects them from. + .add_volume_mount(LOG_VOLUME_NAME.as_ref(), STACKABLE_LOG_DIR) + .context(AddVolumeMountSnafu)? .resources(sidecar_resource_requirements()); add_stackable_rust_cli_env_vars( &mut cb_user_info_fetcher, diff --git a/rust/resource-info-fetcher/Cargo.toml b/rust/resource-info-fetcher/Cargo.toml index d950c78e..2b1cd116 100644 --- a/rust/resource-info-fetcher/Cargo.toml +++ b/rust/resource-info-fetcher/Cargo.toml @@ -29,6 +29,7 @@ url.workspace = true [dev-dependencies] rstest.workspace = true +wiremock.workspace = true [build-dependencies] built.workspace = true diff --git a/rust/resource-info-fetcher/src/api.rs b/rust/resource-info-fetcher/src/api.rs index b270c2ed..e4d873d4 100644 --- a/rust/resource-info-fetcher/src/api.rs +++ b/rust/resource-info-fetcher/src/api.rs @@ -1,3 +1,5 @@ +use std::fmt; + use hyper::StatusCode; use info_fetcher_commons::http_error; use serde::{Deserialize, Serialize}; @@ -38,56 +40,122 @@ pub enum ResourceInfoRequest { RawIdentifier(RawIdentifier), } +/// The maximum length, in bytes, of a single query parameter value. +/// +/// Every parameter value ends up in the response cache key, so without a bound any caller who can +/// name a resource can fill the cache with arbitrarily large keys. Real identifiers are nowhere near +/// this: even a fully qualified DataHub URN stays in the low hundreds of bytes. +const MAX_PARAM_VALUE_LENGTH: usize = 1024; + +/// A query parameter value, bounded to [`MAX_PARAM_VALUE_LENGTH`] bytes. +/// +/// The bound is enforced while deserializing, so an over-long value is reported through the same +/// `400` envelope as any other malformed parameter (see [`crate::MetadataQuery`]) and never reaches +/// the backend or the cache. +/// +/// Serializes as a plain string, so the container key hashes in +/// [`urn_for_request`](crate::backend::data_hub::resource_to_urn_mapping::urn_for_request) are +/// unaffected by the wrapper. +#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)] +pub struct ParamValue(String); + +impl<'de> Deserialize<'de> for ParamValue { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + let value = String::deserialize(deserializer)?; + + if value.len() > MAX_PARAM_VALUE_LENGTH { + return Err(serde::de::Error::custom(format!( + "value is {length} bytes long, but at most {MAX_PARAM_VALUE_LENGTH} are allowed", + length = value.len(), + ))); + } + + Ok(Self(value)) + } +} + +impl fmt::Display for ParamValue { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&self.0) + } +} + +impl AsRef for ParamValue { + fn as_ref(&self) -> &str { + &self.0 + } +} + +/// Only available in tests: outside of them a [`ParamValue`] is always deserialized from a request, +/// which is where the length bound has to be enforced. +#[cfg(test)] +impl From<&str> for ParamValue { + fn from(value: &str) -> Self { + Self(value.to_owned()) + } +} + #[derive(Debug, Clone, PartialEq, Eq, Hash, Deserialize)] pub struct Database { - pub system: String, - pub instance: String, - pub database: String, + pub system: ParamValue, + pub instance: ParamValue, + pub database: ParamValue, } #[derive(Debug, Clone, PartialEq, Eq, Hash, Deserialize)] pub struct Schema { - pub system: String, - pub instance: String, - pub database: String, - pub schema: String, + pub system: ParamValue, + pub instance: ParamValue, + pub database: ParamValue, + pub schema: ParamValue, } #[derive(Debug, Clone, PartialEq, Eq, Hash, Deserialize)] pub struct Table { - pub system: String, - pub instance: String, - pub database: String, - pub schema: String, - pub table: String, + pub system: ParamValue, + pub instance: ParamValue, + pub database: ParamValue, + pub schema: ParamValue, + pub table: ParamValue, } #[derive(Debug, Clone, PartialEq, Eq, Hash, Deserialize)] pub struct Stream { - pub system: String, - pub instance: String, + pub system: ParamValue, + pub instance: ParamValue, /// AKA topic - pub queue: String, + pub queue: ParamValue, } #[derive(Debug, Clone, PartialEq, Eq, Hash, Deserialize)] pub struct Dashboard { - pub system: String, - pub instance: String, - pub id: u64, + pub system: ParamValue, + pub instance: ParamValue, + + /// The dashboard's identifier within its product, treated as an opaque string. + /// + /// Superset numbers its dashboards, but other products (e.g. Looker or Tableau) identify them by + /// name, so this must not be narrowed to an integer. It is only ever spliced into the URN. + pub id: ParamValue, } #[derive(Debug, Clone, PartialEq, Eq, Hash, Deserialize)] pub struct Chart { - pub system: String, - pub instance: String, - pub id: u64, + pub system: ParamValue, + pub instance: ParamValue, + + /// The chart's identifier within its product, treated as an opaque string. See + /// [`Dashboard::id`]. + pub id: ParamValue, } #[derive(Debug, Clone, PartialEq, Eq, Hash, Deserialize)] pub struct RawIdentifier { - pub identifier: String, + pub identifier: ParamValue, } /// Generates the trivial `From for ResourceInfoRequest` conversions, so each HTTP handler @@ -128,16 +196,61 @@ pub enum GetResourceInfoError { } impl http_error::Error for GetResourceInfoError { + /// The status code also decides at which level a failure is logged, see `fetch_resource_info`. + /// Should the client/server split ever prove too coarse for that, the level can be made a + /// property of the individual error instead. fn status_code(&self) -> StatusCode { - // todo: the warn here loses context about the scope in which the error occurred, eg: stackable_opa_resource_info_fetcher::backend::DATA_HUB - // Also, we should make the log level (warn vs error) more dynamic in the backend's impl `http_error::Error for Error` - tracing::warn!( - error = self as &dyn std::error::Error, - "Error while processing request" - ); match self { Self::SerializeResponseAsJson { .. } => StatusCode::INTERNAL_SERVER_ERROR, Self::DataHub { source } => source.status_code(), } } } + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + + /// Deserializes a [`Table`] whose `table` parameter is `length` bytes long. + fn table_with_name_of_length(length: usize) -> Result { + serde_json::from_value(json!({ + "system": "trino", + "instance": "my-trino", + "database": "tpch", + "schema": "sf1", + "table": "a".repeat(length), + })) + } + + /// Parameter values end up in the response cache key, so an unbounded one lets any caller who can + /// name a table fill the cache with megabyte-sized keys. They are bounded while deserializing, + /// which is the same path that renders a `400` for any other malformed parameter. + #[test] + fn param_values_within_the_limit_are_accepted() { + table_with_name_of_length(MAX_PARAM_VALUE_LENGTH) + .expect("a value at the limit must be accepted"); + } + + #[test] + fn param_values_over_the_limit_are_rejected() { + let error = table_with_name_of_length(MAX_PARAM_VALUE_LENGTH + 1) + .expect_err("a value over the limit must be rejected"); + + assert!( + error.to_string().contains("1025 bytes"), + "the error should report the offending length, but was: {error}" + ); + } + + /// The wrapper must stay invisible in the serialized form, because the container URNs are MD5 + /// hashes over the serialized parameters and have to keep matching DataHub's own hashes. + #[test] + fn param_values_serialize_as_plain_strings() { + let serialized = serde_json::to_string(&ParamValue::from("tpch")) + .expect("a param value must be serializable"); + + assert_eq!(serialized, r#""tpch""#); + } +} diff --git a/rust/resource-info-fetcher/src/backend/data_hub/graphql.rs b/rust/resource-info-fetcher/src/backend/data_hub/graphql.rs index d4d77ac2..407ab06a 100644 --- a/rust/resource-info-fetcher/src/backend/data_hub/graphql.rs +++ b/rust/resource-info-fetcher/src/backend/data_hub/graphql.rs @@ -13,11 +13,18 @@ use crate::backend::data_hub::{ }; /// The page size of the `DataProductContains` relationship query, passed to DataHub as the -/// `$dataProductsCount` variable. DataHub paginates the `relationships` resolver, so some page size -/// has to be picked; an asset normally belongs to a single data product, which leaves ample -/// headroom. If it is ever exceeded we fail the request instead of answering with a truncated list, -/// see [`Entity::data_products_truncation`]. -const DATA_PRODUCTS_PAGE_SIZE: u32 = 10; +/// `$dataProductsCount` variable. +/// +/// DataHub paginates the `relationships` resolver, so some page size has to be picked. An asset +/// normally belongs to one or two data products, so this is deliberately set far above anything +/// realistic rather than at a plausible maximum: exceeding it fails the request (see +/// [`Entity::data_products_truncation`]), and failing a lookup that should have succeeded is the +/// worse outcome. It stays a single request at any size, and only costs DataHub more when an asset +/// really does have that many relationships. +/// +/// We do not paginate. That would mean one round trip per page for a case that should not occur, +/// whereas overshooting the page size costs nothing until it is actually needed. +const DATA_PRODUCTS_PAGE_SIZE: u32 = 1000; /// A single query covering every entity kind we build URNs for. We use the generic `entity(urn:)` /// resolver plus per-type inline fragments, because a request can target a dataset (Trino table or @@ -29,6 +36,11 @@ query ResourceInfo($urn: String!, $dataProductsCount: Int!) { } } fragment ResourceInfo on Entity { + # The concrete type DataHub resolved the URN to. Only the types with an inline fragment below carry + # tags, owners and a domain in our response, so this lets us tell an entity we cannot read from one + # that genuinely has no metadata, see `Entity::uncovered_type`. + __typename + # DataHub has no direct "dataProduct" field on assets; membership is a graph edge that points from # the data product to its assets. From the asset's side it is therefore an INCOMING relationship. # `total` is the number of edges DataHub has, which we compare against the number we received to @@ -119,12 +131,24 @@ pub struct ResponseData { #[derive(Debug, Default, Deserialize)] #[serde(rename_all = "camelCase")] pub struct Entity { + /// The concrete DataHub type the URN resolved to, e.g. `Dataset`. [`None`] for the substituted + /// "no metadata" entity, and for a DataHub that does not report it. + #[serde(rename = "__typename")] + typename: Option, + tags: Option, ownership: Option, domain: Option, data_products: Option, } +/// The entity types [`RESOURCE_INFO_QUERY`] has an inline fragment for, and whose tags, owners and +/// domain we therefore read. +/// +/// Anything else still resolves, because `rawIdentifier` accepts any URN, but only the fields common +/// to every `Entity` come back. The response then looks just like that of a resource with no metadata. +const COVERED_ENTITY_TYPES: &[&str] = &["Dataset", "Container", "Chart", "Dashboard"]; + #[derive(Debug, Deserialize)] struct GlobalTags { tags: Vec, @@ -263,6 +287,19 @@ pub struct DataProductsTruncation { } impl Entity { + /// The entity's DataHub type, if [`RESOURCE_INFO_QUERY`] does not cover it. + /// + /// [`None`] means the type is covered, or that DataHub did not report one. In the latter case + /// there is nothing to compare against, so we must not report a problem we cannot substantiate. + /// + /// Callers should surface this: the response for an uncovered type is empty, and a policy has no + /// way to distinguish that from a resource that carries no tags, owners or domain at all. + pub fn uncovered_type(&self) -> Option<&str> { + let typename = self.typename.as_deref()?; + + (!COVERED_ENTITY_TYPES.contains(&typename)).then_some(typename) + } + /// Checks whether the data product list was truncated by [`DATA_PRODUCTS_PAGE_SIZE`]. /// /// Callers must turn this into an error rather than serving the truncated list: a policy that @@ -457,6 +494,43 @@ mod tests { assert!(Entity::default().data_products_truncation().is_none()); } + /// Deserializes an entity of the given DataHub type, as reported by `__typename`. + fn entity_of_type(typename: Option<&str>) -> Entity { + serde_json::from_value(json!({"__typename": typename})) + .expect("test entity must be a valid GraphQL entity payload") + } + + /// An entity type the query has an inline fragment for is read normally. + #[rstest] + #[case::dataset("Dataset")] + #[case::container("Container")] + #[case::chart("Chart")] + #[case::dashboard("Dashboard")] + fn covered_entity_types(#[case] typename: &str) { + assert_eq!(entity_of_type(Some(typename)).uncovered_type(), None); + } + + /// Any other type deserializes into an entity with no tags, owners or domain, which a policy + /// cannot tell apart from a resource that genuinely has none, so it has to be reported. + #[rstest] + #[case::data_job("DataJob")] + #[case::data_flow("DataFlow")] + #[case::notebook("Notebook")] + #[case::ml_model("MLModel")] + fn uncovered_entity_types(#[case] typename: &str) { + assert_eq!( + entity_of_type(Some(typename)).uncovered_type(), + Some(typename) + ); + } + + /// Without a `__typename` there is nothing to check against, so we must not cry wolf. + #[test] + fn entities_without_a_typename_are_not_reported() { + assert_eq!(entity_of_type(None).uncovered_type(), None); + assert_eq!(Entity::default().uncovered_type(), None); + } + #[test] fn truncated_data_products_are_detected() { let truncation = entity(Some(DATA_PRODUCTS_PAGE_SIZE + 1), DATA_PRODUCTS_PAGE_SIZE) @@ -466,4 +540,221 @@ mod tests { assert_eq!(truncation.total, DATA_PRODUCTS_PAGE_SIZE + 1); assert_eq!(truncation.received, DATA_PRODUCTS_PAGE_SIZE); } + + fn urn() -> Urn { + Urn("urn:li:dataset:(urn:li:dataPlatform:trino,my-trino.tpch.sf1.customer,PROD)".to_owned()) + } + + /// Deserializes an entity from the payload DataHub would return. + fn entity_from(payload: serde_json::Value) -> Entity { + serde_json::from_value(payload).expect("test entity must be a valid GraphQL entity payload") + } + + /// The mapping when DataHub populated every `properties` aspect, i.e. the case the fallbacks below + /// are fallbacks for. + #[test] + fn a_fully_populated_entity_maps_every_field() { + let response = entity_from(json!({ + "__typename": "Dataset", + "tags": {"tags": [{"tag": {"urn": "urn:li:tag:PII", "properties": {"name": "PII"}}}]}, + "domain": {"domain": {"urn": "urn:li:domain:finance", "properties": { + "name": "Finance", "description": "Financial data", + }}}, + "dataProducts": {"total": 1, "relationships": [{"entity": { + "urn": "urn:li:dataProduct:orders", + "properties": {"name": "Orders", "description": "Order data"}, + }}]}, + "ownership": {"owners": [ + { + "owner": { + "__typename": "CorpUser", + "urn": "urn:li:corpuser:alice", + "properties": { + "fullName": "Alice Example", "displayName": "Alice", + "email": "alice@example.com", "active": true, + }, + }, + "ownershipType": { + "urn": "urn:li:ownershipType:__system__technical_owner", + "info": {"name": "Technical Owner"}, + }, + }, + { + "owner": { + "__typename": "CorpGroup", + "urn": "urn:li:corpGroup:analytics", + "properties": {"displayName": "Analytics", "description": "The team"}, + }, + "ownershipType": { + "urn": "urn:li:ownershipType:__system__technical_owner", + "info": {"name": "Technical Owner"}, + }, + }, + ]}, + })) + .into_response(urn()); + + assert_eq!( + response.tags, + vec![Tag { + urn: Urn("urn:li:tag:PII".to_owned()), + name: "PII".to_owned(), + }] + ); + assert_eq!( + response.domain, + Some(Domain { + urn: Urn("urn:li:domain:finance".to_owned()), + name: "Finance".to_owned(), + description: Some("Financial data".to_owned()), + }) + ); + assert_eq!( + response.data_products, + vec![DataProduct { + urn: Urn("urn:li:dataProduct:orders".to_owned()), + name: "Orders".to_owned(), + description: Some("Order data".to_owned()), + }] + ); + + // Both owners share an ownership type, so they land in the same bucket rather than two. + let technical_owner = Urn("urn:li:ownershipType:__system__technical_owner".to_owned()); + assert_eq!(response.owners.len(), 1); + let owners = &response.owners[&technical_owner]; + assert_eq!( + owners.ownership_type_name.as_deref(), + Some("Technical Owner") + ); + assert_eq!( + owners.users, + vec![User { + urn: Urn("urn:li:corpuser:alice".to_owned()), + full_name: Some("Alice Example".to_owned()), + display_name: "Alice".to_owned(), + email: Some("alice@example.com".to_owned()), + active: true, + }] + ); + assert_eq!( + owners.groups, + vec![Group { + urn: Urn("urn:li:corpGroup:analytics".to_owned()), + display_name: "Analytics".to_owned(), + description: Some("The team".to_owned()), + }] + ); + } + + /// An entity whose referenced tag, domain and data product have no `properties` aspect. There is no + /// name to show, so each falls back to its URN rather than to an empty string, which would render + /// as a nameless entry in a policy decision. + #[test] + fn entities_without_properties_fall_back_to_urns() { + let response = entity_from(json!({ + "tags": {"tags": [{"tag": {"urn": "urn:li:tag:PII"}}]}, + "domain": {"domain": {"urn": "urn:li:domain:finance"}}, + "dataProducts": { + "total": 1, + "relationships": [{"entity": {"urn": "urn:li:dataProduct:orders"}}], + }, + })) + .into_response(urn()); + + assert_eq!(response.tags[0].name, "urn:li:tag:PII"); + + let domain = response.domain.expect("the domain is present"); + assert_eq!(domain.name, "urn:li:domain:finance"); + assert_eq!(domain.description, None); + + assert_eq!(response.data_products[0].name, "urn:li:dataProduct:orders"); + assert_eq!(response.data_products[0].description, None); + } + + /// Owners whose `properties` aspect is missing entirely, and owners where it exists but carries no + /// display name. A user's display name is derived from the URN; a group's falls back to the whole + /// URN, as there is no group-name convention to strip. + #[rstest] + #[case::no_properties_aspect(json!({"__typename": "CorpUser", "urn": "urn:li:corpuser:alice"}))] + #[case::no_display_name( + json!({"__typename": "CorpUser", "urn": "urn:li:corpuser:alice", "properties": {}}) + )] + fn users_without_a_display_name_are_named_after_their_urn(#[case] owner: serde_json::Value) { + let response = + entity_from(json!({"ownership": {"owners": [{"owner": owner}]}})).into_response(urn()); + + let owners = response + .owners + .values() + .next() + .expect("the owner is present"); + assert_eq!(owners.users[0].display_name, "alice"); + assert_eq!(owners.users[0].full_name, None); + assert_eq!(owners.users[0].email, None); + // Absent `active` means we must not report the user as deactivated. + assert!(owners.users[0].active); + } + + #[rstest] + #[case::no_properties_aspect( + json!({"__typename": "CorpGroup", "urn": "urn:li:corpGroup:analytics"}) + )] + #[case::no_display_name( + json!({"__typename": "CorpGroup", "urn": "urn:li:corpGroup:analytics", "properties": {}}) + )] + fn groups_without_a_display_name_are_named_after_their_urn(#[case] owner: serde_json::Value) { + let response = + entity_from(json!({"ownership": {"owners": [{"owner": owner}]}})).into_response(urn()); + + let owners = response + .owners + .values() + .next() + .expect("the owner is present"); + assert_eq!(owners.groups[0].display_name, "urn:li:corpGroup:analytics"); + assert_eq!(owners.groups[0].description, None); + } + + /// Owners predating ownership type entities only carry the legacy `type` enum, and some carry + /// neither. The key has to stay stable either way, because it is what a policy looks owners up by. + #[rstest] + #[case::legacy_type_only(json!({"type": "TECHNICAL_OWNER"}), "TECHNICAL_OWNER")] + #[case::no_type_at_all(json!({}), "unknown")] + fn owners_without_an_ownership_type_entity_fall_back_to_the_legacy_type( + #[case] extra_owner_fields: serde_json::Value, + #[case] expected_key: &str, + ) { + let mut owner = json!({ + "owner": {"__typename": "CorpUser", "urn": "urn:li:corpuser:alice"}, + }); + owner + .as_object_mut() + .expect("the owner is a JSON object") + .extend( + extra_owner_fields + .as_object() + .expect("the extra fields are a JSON object") + .clone(), + ); + + let response = entity_from(json!({"ownership": {"owners": [owner]}})).into_response(urn()); + + let (key, owners) = response.owners.iter().next().expect("the owner is present"); + assert_eq!(key.0, expected_key); + // There is no ownership type entity, so there is no human-readable name for it either. + assert_eq!(owners.ownership_type_name, None); + } + + /// The "no metadata" entity we substitute for a URN DataHub does not know must map to a response + /// that is empty rather than one that fails to build. + #[test] + fn the_default_entity_maps_to_an_empty_response() { + let response = Entity::default().into_response(urn()); + + assert_eq!(response.urn, urn()); + assert!(response.tags.is_empty()); + assert_eq!(response.domain, None); + assert!(response.data_products.is_empty()); + assert!(response.owners.is_empty()); + } } diff --git a/rust/resource-info-fetcher/src/backend/data_hub/mod.rs b/rust/resource-info-fetcher/src/backend/data_hub/mod.rs index cd55d79b..372d5795 100644 --- a/rust/resource-info-fetcher/src/backend/data_hub/mod.rs +++ b/rust/resource-info-fetcher/src/backend/data_hub/mod.rs @@ -13,7 +13,7 @@ use reqwest::Url; use serde::{Deserialize, Serialize}; use snafu::{ResultExt, Snafu}; use stackable_opa_operator::crd::resource_info_fetcher::v1alpha1; -use tracing::{debug, instrument, trace}; +use tracing::{debug, instrument, trace, warn}; use crate::{ api::{GetResourceInfoError, ResourceInfoBackend, ResourceInfoRequest}, @@ -21,10 +21,14 @@ use crate::{ }; mod graphql; -mod resource_to_urn_mapping; +pub(crate) mod resource_to_urn_mapping; +/// Errors that can occur while resolving the backend, which happens once at startup. +/// +/// Kept apart from [`Error`] because these never reach a caller: a failure here means the process +/// does not come up at all, so (unlike [`Error`]) they have no HTTP status code to map to. #[derive(Snafu, Debug)] -pub enum Error { +pub enum ResolveError { #[snafu(display("failed to read DataHub token from {path:?}"))] ReadToken { source: std::io::Error, @@ -42,7 +46,12 @@ pub enum Error { source: url::ParseError, endpoint: String, }, +} +/// Errors that can occur while answering a request, and which are therefore rendered as an HTTP +/// response to the caller. +#[derive(Snafu, Debug)] +pub enum Error { #[snafu(display("failed to execute GraphQL query for URN {urn:?}"))] ExecuteGraphQlQuery { source: utils::http::Error, @@ -52,6 +61,13 @@ pub enum Error { #[snafu(display("DataHub returned GraphQL errors for URN {urn:?}: {messages}"))] GraphQlErrors { messages: String, urn: Urn }, + #[snafu(display( + "the resource name {name:?} contains {delimiter:?}, which DataHub uses to delimit the parts \ + of a URN. No URN containing it can resolve, so the request is rejected without querying \ + DataHub." + ))] + InvalidResourceName { name: String, delimiter: char }, + #[snafu(display( "DataHub reported {total} data products for URN {urn:?}, but the GraphQL query only fetches \ up to {received} of them. Refusing to answer with a truncated list of data products, as \ @@ -64,12 +80,18 @@ pub enum Error { impl http_error::Error for Error { fn status_code(&self) -> StatusCode { match self { - Self::ReadToken { .. } => StatusCode::SERVICE_UNAVAILABLE, - Self::ConfigureTls { .. } => StatusCode::SERVICE_UNAVAILABLE, - Self::ConstructHttpClient { .. } => StatusCode::SERVICE_UNAVAILABLE, - Self::BuildDataHubEndpoint { .. } => StatusCode::BAD_REQUEST, + // We could not talk to DataHub at all, which is not something the caller can fix. Self::ExecuteGraphQlQuery { .. } => StatusCode::INTERNAL_SERVER_ERROR, - Self::GraphQlErrors { .. } => StatusCode::INTERNAL_SERVER_ERROR, + + // The URN is built entirely from the caller's parameters, so a URN DataHub refuses to + // parse or resolve is a bad request rather than a server fault. Any user who can name a + // table can reach this, e.g. through Trino's `SELECT * FROM tpch.sf1."a,PROD)"`. + Self::GraphQlErrors { .. } => StatusCode::BAD_REQUEST, + + // The caller named a resource that cannot be expressed as a URN at all. + Self::InvalidResourceName { .. } => StatusCode::BAD_REQUEST, + + // A limitation of our own query, see the variant's message. Self::TruncatedDataProducts { .. } => StatusCode::INTERNAL_SERVER_ERROR, } } @@ -177,7 +199,7 @@ impl ResolvedDataHubBackend { pub async fn resolve( config: v1alpha1::DataHubBackend, credentials_dir: &Path, - ) -> Result { + ) -> Result { let token_path = credentials_dir.join("token"); // Trim trailing whitespace/newlines so the value is safe to use in an HTTP header. @@ -245,6 +267,20 @@ impl ResolvedDataHubBackend { return Ok(graphql::Entity::default()); }; + // The query only reads tags, owners and domains off the entity types it has inline fragments + // for. Anything else (reachable through `rawIdentifier`, which accepts any URN) resolves to + // a response that looks just like that of a resource with no metadata, so say so rather than + // letting a policy silently decide on an empty record. + if let Some(entity_type) = entity.uncovered_type() { + warn!( + %urn, + entity_type, + "DataHub resolved this URN to an entity type the resource-info-fetcher cannot read \ + metadata from; answering with empty tags, owners and data products. A policy cannot \ + tell this apart from a resource that has no metadata, so do not rely on it" + ); + } + // Fail loudly instead of serving a partial list of data products: a policy that keys off data // product membership would otherwise silently decide based on incomplete metadata. if let Some(truncation) = entity.data_products_truncation() { @@ -263,8 +299,22 @@ impl ResolvedDataHubBackend { } } +#[cfg(test)] +impl ResolvedDataHubBackend { + /// A backend querying `graphql_url`, bypassing [`ResolvedDataHubBackend::resolve`] so that no + /// credentials have to be read from disk. + pub fn for_tests(graphql_url: Url) -> Self { + Self { + token: "not-a-real-token".to_owned(), + http_client: reqwest::Client::new(), + graphql_url, + env: v1alpha1::FabricType::Prod, + } + } +} + /// Builds the DataHub GraphQL endpoint from the backend configuration. -fn build_graphql_url(config: &v1alpha1::DataHubBackend) -> Result { +fn build_graphql_url(config: &v1alpha1::DataHubBackend) -> Result { let schema = if config.tls.uses_tls() { "https" } else { @@ -303,7 +353,7 @@ impl ResourceInfoBackend for ResolvedDataHubBackend { &self, request: &ResourceInfoRequest, ) -> Result { - let urn = urn_for_request(request, &self.env); + let urn = urn_for_request(request, &self.env)?; let entity = self.query_entity(&urn).await?; Ok(entity.into_response(urn)) @@ -314,6 +364,7 @@ impl ResourceInfoBackend for ResolvedDataHubBackend { mod tests { use rstest::rstest; use serde_json::json; + use snafu::IntoError; use super::*; @@ -348,6 +399,39 @@ mod tests { /// [`Url`] omits the port whenever it is the default port of the scheme, hence the expected /// endpoints of the defaulted cases carry no port. + fn urn() -> Urn { + Urn("urn:li:dataset:(urn:li:dataPlatform:trino,a,PROD)".to_owned()) + } + + /// The status code tells the caller whose problem a failure is. The URN we query is built + /// entirely from the caller's parameters, so a URN DataHub refuses to parse or resolve is a bad + /// request. Any user who can name a table can reach it, for example through Trino's + /// `SELECT * FROM tpch.sf1."a,PROD)"`. A backend we could not reach at all, or a limitation of + /// our own query, is not something the caller can do anything about. + #[rstest] + #[case::graphql_errors( + GraphQlErrorsSnafu { messages: "Failed to parse urn", urn: urn() }.build(), + StatusCode::BAD_REQUEST + )] + #[case::unreachable_backend( + ExecuteGraphQlQuerySnafu { urn: urn() }.into_error(utils::http::Error::HttpErrorResponse { + status: StatusCode::UNAUTHORIZED, + url: "http://datahub-gms/api/graphql".to_owned(), + text: "Unauthorized".to_owned(), + }), + StatusCode::INTERNAL_SERVER_ERROR + )] + #[case::truncated_data_products( + TruncatedDataProductsSnafu { urn: urn(), total: 11u32, received: 10u32 }.build(), + StatusCode::INTERNAL_SERVER_ERROR + )] + fn status_code(#[case] error: Error, #[case] expected_status_code: StatusCode) { + assert_eq!( + info_fetcher_commons::http_error::Error::status_code(&error), + expected_status_code + ); + } + #[rstest] #[case::default_scheme_and_port(json!({}), format!("http://{HOSTNAME}/api/graphql"))] #[case::default_tls_scheme_and_port( diff --git a/rust/resource-info-fetcher/src/backend/data_hub/resource_to_urn_mapping.rs b/rust/resource-info-fetcher/src/backend/data_hub/resource_to_urn_mapping.rs index f42d6a28..e666164f 100644 --- a/rust/resource-info-fetcher/src/backend/data_hub/resource_to_urn_mapping.rs +++ b/rust/resource-info-fetcher/src/backend/data_hub/resource_to_urn_mapping.rs @@ -4,8 +4,11 @@ use serde::Serialize; use stackable_opa_operator::crd::resource_info_fetcher::v1alpha1; use crate::{ - api::{Chart, Dashboard, Database, RawIdentifier, ResourceInfoRequest, Schema, Stream, Table}, - backend::data_hub::Urn, + api::{ + Chart, Dashboard, Database, ParamValue, RawIdentifier, ResourceInfoRequest, Schema, Stream, + Table, + }, + backend::data_hub::{Error, InvalidResourceNameSnafu, Urn}, }; /// Maps a request to the URN of the DataHub entity that holds the resource's metadata. @@ -13,28 +16,37 @@ use crate::{ /// `env` is DataHub's fabric (e.g. `PROD`) and comes from the backend configuration rather than from /// the request - see [`v1alpha1::DataHubBackend::env`] for why. It is part of every dataset URN, so /// it has to match the `env` the metadata was ingested with, otherwise the URN does not resolve. -pub fn urn_for_request(request: &ResourceInfoRequest, env: &v1alpha1::FabricType) -> Urn { +pub fn urn_for_request( + request: &ResourceInfoRequest, + env: &v1alpha1::FabricType, +) -> Result { let urn = match request { ResourceInfoRequest::Database(Database { system, instance, database, - }) => container_urn(&BTreeMap::from([ - ("platform", system), - ("instance", instance), - ("database", database), - ])), + }) => { + reject_urn_delimiters(&[system, instance, database])?; + container_urn(&BTreeMap::from([ + ("platform", system), + ("instance", instance), + ("database", database), + ])) + } ResourceInfoRequest::Schema(Schema { system, instance, database, schema, - }) => container_urn(&BTreeMap::from([ - ("platform", system), - ("instance", instance), - ("database", database), - ("schema", schema), - ])), + }) => { + reject_urn_delimiters(&[system, instance, database, schema])?; + container_urn(&BTreeMap::from([ + ("platform", system), + ("instance", instance), + ("database", database), + ("schema", schema), + ])) + } ResourceInfoRequest::Table(Table { system, instance, @@ -42,6 +54,7 @@ pub fn urn_for_request(request: &ResourceInfoRequest, env: &v1alpha1::FabricType schema, table, }) => { + reject_urn_delimiters(&[system, instance, database, schema, table])?; format!( "urn:li:dataset:(urn:li:dataPlatform:{system},{instance}.{database}.{schema}.{table},{env})" ) @@ -50,12 +63,16 @@ pub fn urn_for_request(request: &ResourceInfoRequest, env: &v1alpha1::FabricType system, instance, queue, - }) => format!("urn:li:dataset:(urn:li:dataPlatform:{system},{instance}.{queue},{env})"), + }) => { + reject_urn_delimiters(&[system, instance, queue])?; + format!("urn:li:dataset:(urn:li:dataPlatform:{system},{instance}.{queue},{env})") + } ResourceInfoRequest::Dashboard(Dashboard { system, instance, id, }) => { + reject_urn_delimiters(&[system, instance, id])?; format!("urn:li:dashboard:({system},{instance}.{id})") } ResourceInfoRequest::Chart(Chart { @@ -63,12 +80,42 @@ pub fn urn_for_request(request: &ResourceInfoRequest, env: &v1alpha1::FabricType instance, id, }) => { + reject_urn_delimiters(&[system, instance, id])?; format!("urn:li:chart:({system},{instance}.{id})") } - ResourceInfoRequest::RawIdentifier(RawIdentifier { identifier }) => identifier.clone(), + // Deliberately unchecked: a raw identifier *is* a URN, so it necessarily contains the + // delimiters that are rejected above. + ResourceInfoRequest::RawIdentifier(RawIdentifier { identifier }) => identifier.to_string(), }; - Urn(urn) + Ok(Urn(urn)) +} + +/// The characters DataHub's URN grammar uses to delimit the parts of a URN. +/// +/// A resource name containing one of these cannot be expressed as a URN at all, because DataHub's own +/// parser would split the name apart. No URN we build for it would ever resolve. +const URN_DELIMITERS: [char; 3] = [',', '(', ')']; + +/// Fails if any of `names` contains a [`URN_DELIMITERS`] character. +/// +/// This is checked before querying DataHub rather than after: the query is guaranteed to fail, and any +/// caller who can name a resource could otherwise turn every such name into a round trip to DataHub +/// plus a log line. For a Trino table, `SELECT * FROM tpch.sf1."a,PROD)"` is enough. +fn reject_urn_delimiters(names: &[&ParamValue]) -> Result<(), Error> { + for name in names { + if let Some(delimiter) = name.as_ref().find(URN_DELIMITERS) { + let name = name.to_string(); + let delimiter = name[delimiter..] + .chars() + .next() + .expect("the match starts at a character boundary"); + + return InvalidResourceNameSnafu { name, delimiter }.fail(); + } + } + + Ok(()) } /// Reproduces DataHub's `datahub_guid`: the container key is serialized to compact, key-sorted JSON @@ -87,3 +134,185 @@ fn container_urn( .expect("serializing a BTreeMap<&str, &str> cannot fail"); format!("urn:li:container:{:x}", md5::compute(key_json.as_bytes())) } + +#[cfg(test)] +mod tests { + use hyper::StatusCode; + use rstest::rstest; + + use super::*; + + /// Dashboard and chart ids are opaque to us. Superset happens to number them, but other + /// platforms (e.g. Looker or Tableau) identify their dashboards by name, so neither the API nor + /// the URN construction may assume an integer. + /// + /// Neither URN carries the fabric, so the configured [`v1alpha1::FabricType`] is irrelevant here. + #[rstest] + #[case::numeric_id("1", "urn:li:chart:(superset,my-superset.1)")] + #[case::non_numeric_id( + "orders-by-region", + "urn:li:chart:(superset,my-superset.orders-by-region)" + )] + fn chart_urn(#[case] id: &str, #[case] expected_urn: &str) { + let request = ResourceInfoRequest::Chart(Chart { + system: "superset".into(), + instance: "my-superset".into(), + id: id.into(), + }); + + assert_eq!(urn_of(request).expect("the name is valid").0, expected_urn); + } + + fn urn_of(request: ResourceInfoRequest) -> Result { + urn_for_request(&request, &v1alpha1::FabricType::Prod) + } + + fn table_named(table: &str) -> ResourceInfoRequest { + ResourceInfoRequest::Table(Table { + system: "trino".into(), + instance: "my-trino".into(), + database: "tpch".into(), + schema: "sf1".into(), + table: table.into(), + }) + } + + /// A name containing a URN delimiter cannot be expressed as a DataHub URN at all, so querying + /// DataHub with it is guaranteed to fail. Rejecting it here stops a caller who can name a table + /// (via Trino's `SELECT * FROM tpch.sf1."a,PROD)"`, say) from turning every such query into a + /// round trip to DataHub. + #[rstest] + #[case::comma("a,PROD)")] + #[case::opening_paren("a(b")] + #[case::closing_paren("a)b")] + fn names_containing_urn_delimiters_are_rejected(#[case] table: &str) { + let error = urn_of(table_named(table)) + .expect_err("a name containing a URN delimiter must be rejected"); + + assert_eq!( + info_fetcher_commons::http_error::Error::status_code(&error), + StatusCode::BAD_REQUEST + ); + } + + /// Dots are not delimiters: they separate the segments of the dataset name, and a name that + /// genuinely contains one resolves as long as it was ingested the same way. + #[rstest] + #[case::plain("customer")] + #[case::dotted("a.b")] + #[case::dashed_and_underscored("my_table-2")] + #[case::colon("a:b")] + fn ordinary_names_are_accepted(#[case] table: &str) { + urn_of(table_named(table)).expect("an ordinary name must be accepted"); + } + + /// `rawIdentifier` is passed through verbatim, and for DataHub it *is* a URN, so it necessarily + /// contains the very delimiters the other endpoints reject. + #[test] + fn raw_identifiers_may_contain_urn_delimiters() { + let identifier = "urn:li:chart:(superset,my-superset.1)"; + + let urn = urn_of(ResourceInfoRequest::RawIdentifier(RawIdentifier { + identifier: identifier.into(), + })) + .expect("a raw identifier must be passed through as-is"); + + assert_eq!(urn.0, identifier); + } + + /// Guards the container hashes, which DataHub computes independently on ingestion: they have to + /// keep matching byte for byte, or every database and schema lookup silently stops resolving. + /// + /// The expected values were computed with Python's + /// `json.dumps(key, sort_keys=True, separators=(",", ":"))` and `hashlib.md5`, mirroring what + /// DataHub's `datahub_guid` does. They were not read back out of this implementation. + #[rstest] + #[case::database( + ResourceInfoRequest::Database(Database { + system: "trino".into(), + instance: "my-namespace/my-trino".into(), + database: "tpch".into(), + }), + "urn:li:container:9c4275840ebdb59d3b21b4b4102e8999" + )] + #[case::schema( + ResourceInfoRequest::Schema(Schema { + system: "trino".into(), + instance: "my-namespace/my-trino".into(), + database: "tpch".into(), + schema: "sf1".into(), + }), + "urn:li:container:fb46bf1f985e130eeceeee8a51317cd9" + )] + fn container_urns_match_datahubs_guid( + #[case] request: ResourceInfoRequest, + #[case] expected_urn: &str, + ) { + assert_eq!(urn_of(request).expect("the name is valid").0, expected_urn); + } + + /// The dataset URNs are built by string interpolation rather than hashed, so these pin the exact + /// layout, including that the fabric is appended and that a table's four name segments are + /// dot-joined in order. + #[rstest] + #[case::table( + ResourceInfoRequest::Table(Table { + system: "trino".into(), + instance: "my-namespace/my-trino".into(), + database: "tpch".into(), + schema: "sf1".into(), + table: "customer".into(), + }), + "urn:li:dataset:(urn:li:dataPlatform:trino,my-namespace/my-trino.tpch.sf1.customer,PROD)" + )] + #[case::stream( + ResourceInfoRequest::Stream(Stream { + system: "kafka".into(), + instance: "my-namespace/my-kafka".into(), + queue: "orders".into(), + }), + "urn:li:dataset:(urn:li:dataPlatform:kafka,my-namespace/my-kafka.orders,PROD)" + )] + #[case::dashboard( + ResourceInfoRequest::Dashboard(Dashboard { + system: "superset".into(), + instance: "my-namespace/my-superset".into(), + id: "1".into(), + }), + "urn:li:dashboard:(superset,my-namespace/my-superset.1)" + )] + fn interpolated_urns(#[case] request: ResourceInfoRequest, #[case] expected_urn: &str) { + assert_eq!(urn_of(request).expect("the name is valid").0, expected_urn); + } + + /// The fabric is part of every dataset URN, so a mismatch with the ingestion's `env` is what makes + /// an otherwise correct lookup return nothing. + #[test] + fn the_configured_fabric_ends_up_in_dataset_urns() { + let request = ResourceInfoRequest::Stream(Stream { + system: "kafka".into(), + instance: "my-kafka".into(), + queue: "orders".into(), + }); + + let urn = urn_for_request(&request, &v1alpha1::FabricType::Dev).expect("the name is valid"); + + assert_eq!( + urn.0, + "urn:li:dataset:(urn:li:dataPlatform:kafka,my-kafka.orders,DEV)" + ); + } + + #[rstest] + #[case::numeric_id("1", "urn:li:dashboard:(superset,my-superset.1)")] + #[case::non_numeric_id("sales", "urn:li:dashboard:(superset,my-superset.sales)")] + fn dashboard_urn(#[case] id: &str, #[case] expected_urn: &str) { + let request = ResourceInfoRequest::Dashboard(Dashboard { + system: "superset".into(), + instance: "my-superset".into(), + id: id.into(), + }); + + assert_eq!(urn_of(request).expect("the name is valid").0, expected_urn); + } +} diff --git a/rust/resource-info-fetcher/src/cache.rs b/rust/resource-info-fetcher/src/cache.rs new file mode 100644 index 00000000..4832882c --- /dev/null +++ b/rust/resource-info-fetcher/src/cache.rs @@ -0,0 +1,191 @@ +//! The response cache, including the short-lived caching of failed lookups. + +use std::{sync::Arc, time::Duration}; + +use moka::{Expiry, future::Cache}; +use stackable_opa_operator::crd::cache; + +use crate::api::{GetResourceInfoError, ResourceInfoRequest}; + +/// How long a failed lookup is remembered. +/// +/// Without this, a lookup that keeps failing queries the backend again on every single request. That +/// is reachable by anyone who can name a resource, and it is at its worst exactly when the backend is +/// already in trouble. +/// +/// Deliberately far shorter than the configured time-to-live for successful lookups: a stale success +/// is merely out of date, while a stale failure keeps denying (or, for a rule keyed on the absence of +/// a tag, keeps granting) after the backend has recovered. If the configured time-to-live is shorter +/// than this, it wins, because moka evicts at the earliest of the two. +const FAILURE_TIME_TO_LIVE: Duration = Duration::from_secs(5); + +/// What a lookup produced, as held in the cache. +/// +/// Failures are cached as well as successes, so the variants share one key space and one capacity +/// limit, and so that moka coalesces concurrent lookups of a failing key just as it does a +/// successful one. +#[derive(Clone)] +pub enum CachedResponse { + /// The metadata to answer with, already serialized to the JSON we return. + Found(serde_json::Value), + + /// The lookup failed. Held in an [`Arc`] because every caller that hits this entry is handed the + /// same error. + Failed(Arc), +} + +pub type ResourceInfoCache = Cache; + +/// Builds the response cache from the cluster's cache configuration. +pub fn build(config: &cache::Cache) -> ResourceInfoCache { + build_with_failure_time_to_live(config, FAILURE_TIME_TO_LIVE) +} + +fn build_with_failure_time_to_live( + config: &cache::Cache, + failure_time_to_live: Duration, +) -> ResourceInfoCache { + config + .apply_settings_to_cache_builder(Cache::builder().name("resource-info")) + .expire_after(FailureExpiry { + failure_time_to_live, + }) + .build() +} + +/// Expires [`CachedResponse::Failed`] entries early, leaving successful ones to the configured +/// time-to-live. +struct FailureExpiry { + failure_time_to_live: Duration, +} + +impl FailureExpiry { + /// [`None`] leaves the entry to the cache's `time_to_live`, which is what successful lookups get. + fn time_to_live_for(&self, response: &CachedResponse) -> Option { + match response { + CachedResponse::Found(_) => None, + CachedResponse::Failed(_) => Some(self.failure_time_to_live), + } + } +} + +impl Expiry for FailureExpiry { + fn expire_after_create( + &self, + _key: &ResourceInfoRequest, + response: &CachedResponse, + _created_at: std::time::Instant, + ) -> Option { + self.time_to_live_for(response) + } + + fn expire_after_update( + &self, + _key: &ResourceInfoRequest, + response: &CachedResponse, + _updated_at: std::time::Instant, + _duration_until_expiry: Option, + ) -> Option { + self.time_to_live_for(response) + } +} + +#[cfg(test)] +mod tests { + use std::sync::atomic::{AtomicUsize, Ordering}; + + use serde_json::json; + + use super::*; + use crate::api::RawIdentifier; + + /// A cache whose failed entries live for `failure_time_to_live`, and whose successful ones live + /// for a time-to-live long enough not to interfere with any test. + fn cache(failure_time_to_live: Duration) -> ResourceInfoCache { + let config = serde_json::from_value(json!({"entryTimeToLive": "10m"})) + .expect("the cache config must be valid"); + + build_with_failure_time_to_live(&config, failure_time_to_live) + } + + fn request(identifier: &str) -> ResourceInfoRequest { + ResourceInfoRequest::RawIdentifier(RawIdentifier { + identifier: identifier.into(), + }) + } + + fn failure() -> CachedResponse { + let error = GetResourceInfoError::SerializeResponseAsJson { + source: serde_json::from_str::("not json") + .expect_err("the input is not valid JSON"), + }; + + CachedResponse::Failed(Arc::new(error)) + } + + /// Counts how often the cache had to load a value, so the tests assert on cache hits rather than + /// on timing. + #[derive(Default)] + struct Loads(AtomicUsize); + + impl Loads { + async fn load(&self, response: CachedResponse) -> CachedResponse { + self.0.fetch_add(1, Ordering::SeqCst); + response + } + + fn count(&self) -> usize { + self.0.load(Ordering::SeqCst) + } + } + + /// The point of the whole module: a failing lookup must not reach the backend again on the next + /// request. + #[tokio::test] + async fn a_failed_lookup_is_served_from_the_cache() { + let cache = cache(Duration::from_secs(60)); + let loads = Loads::default(); + let request = request("urn:li:dataset:(urn:li:dataPlatform:trino,broken,PROD)"); + + for _ in 0..3 { + cache.get_with_by_ref(&request, loads.load(failure())).await; + } + + assert_eq!(loads.count(), 1); + } + + /// A failure must not be held for the full time-to-live of a successful lookup: the backend may + /// have recovered in the meantime, and until the entry goes away every request keeps failing. + #[tokio::test] + async fn a_failed_lookup_is_forgotten_again_quickly() { + let cache = cache(Duration::from_millis(50)); + let loads = Loads::default(); + let request = request("urn:li:dataset:(urn:li:dataPlatform:trino,broken,PROD)"); + + cache.get_with_by_ref(&request, loads.load(failure())).await; + tokio::time::sleep(Duration::from_millis(100)).await; + cache.run_pending_tasks().await; + cache.get_with_by_ref(&request, loads.load(failure())).await; + + assert_eq!(loads.count(), 2); + } + + /// The short expiry must apply to failures only. A successful lookup keeps the configured + /// time-to-live, which the test's cache sets far beyond the failure one. + #[tokio::test] + async fn a_successful_lookup_keeps_the_configured_time_to_live() { + let cache = cache(Duration::from_millis(50)); + let loads = Loads::default(); + let request = request("urn:li:container:fb46bf1f985e130eeceeee8a51317cd9"); + let found = CachedResponse::Found(json!({"tags": []})); + + cache + .get_with_by_ref(&request, loads.load(found.clone())) + .await; + tokio::time::sleep(Duration::from_millis(100)).await; + cache.run_pending_tasks().await; + cache.get_with_by_ref(&request, loads.load(found)).await; + + assert_eq!(loads.count(), 1); + } +} diff --git a/rust/resource-info-fetcher/src/main.rs b/rust/resource-info-fetcher/src/main.rs index 6bbf79d8..46a1e460 100644 --- a/rust/resource-info-fetcher/src/main.rs +++ b/rust/resource-info-fetcher/src/main.rs @@ -16,17 +16,20 @@ use info_fetcher_commons::{ config::{ConfigError, read_config_file}, http_error, }; -use moka::future::Cache; use serde::de::DeserializeOwned; use snafu::{ResultExt, Snafu}; use stackable_opa_operator::crd::resource_info_fetcher::v1alpha1::{self}; use stackable_operator::{cli::CommonOptions, telemetry::Tracing}; use tokio::net::TcpListener; -use crate::api::{GetResourceInfoError, ResourceInfoBackend, ResourceInfoRequest}; +use crate::{ + api::{GetResourceInfoError, ResourceInfoBackend, ResourceInfoRequest}, + cache::{CachedResponse, ResourceInfoCache}, +}; mod api; mod backend; +mod cache; pub mod built_info { include!(concat!(env!("OUT_DIR"), "/built.rs")); @@ -51,7 +54,7 @@ struct AppState { backend: Arc, // Note: Although we might not talk JSON to the underlying backend, we always return JSON as a // result to the caller, so we can cache that. - resource_info_cache: Cache, + resource_info_cache: ResourceInfoCache, } /// Backend with resolved credentials. @@ -82,7 +85,9 @@ enum StartupError { RunServer { source: std::io::Error }, #[snafu(display("failed to resolve DataHub backend"))] - ResolveDataHubBackend { source: backend::data_hub::Error }, + ResolveDataHubBackend { + source: backend::data_hub::ResolveError, + }, } /// Resolves a backend configuration by loading credentials and creating the appropriate backend implementation. @@ -139,10 +144,7 @@ async fn main() -> Result<(), StartupError> { let config: v1alpha1::Config = read_config_file(&args.config) .with_context(|_| ParseConfigFileSnafu { path: args.config })?; let backend = Arc::new(resolve_backend(config.backend, &args.credentials_dir).await?); - let resource_info_cache = config - .cache - .apply_settings_to_cache_builder(Cache::builder().name("resource-info")) - .build(); + let resource_info_cache = cache::build(&config.cache); // One GET endpoint per resource type. They all share the same generic `metadata` handler; only // the query-parameter struct (and thus the resulting `ResourceInfoRequest` variant) differs. let app = Router::new() @@ -232,18 +234,106 @@ async fn get_resource_info( backend, resource_info_cache, } = state; - let resource_info = resource_info_cache - .try_get_with_by_ref(&request, async { - match backend.as_ref() { - ResolvedBackend::DataHub(data_hub) => { - let response = data_hub.get_resource_info(&request).await?; - serde_json::to_value(&response).map_err(|err| { - GetResourceInfoError::SerializeResponseAsJson { source: err } - }) - } + // A failed lookup is cached as well, see [`CachedResponse`], so nothing fails here: the error is + // part of the cached value rather than something the cache passes through. + let cached = resource_info_cache + .get_with_by_ref(&request, async { + match fetch_resource_info(&backend, &request).await { + Ok(resource_info) => CachedResponse::Found(resource_info), + Err(error) => CachedResponse::Failed(Arc::new(error)), } }) - .await?; + .await; + + match cached { + CachedResponse::Found(resource_info) => Ok(Json(resource_info)), + CachedResponse::Failed(error) => Err(error.into()), + } +} + +/// Queries the backend for a single resource and serializes the answer into the JSON we return. +async fn fetch_resource_info( + backend: &ResolvedBackend, + request: &ResourceInfoRequest, +) -> Result { + let resource_info = + match backend { + ResolvedBackend::DataHub(data_hub) => data_hub + .get_resource_info(request) + .await + .and_then(|response| { + serde_json::to_value(&response) + .map_err(|source| GetResourceInfoError::SerializeResponseAsJson { source }) + }), + }; + + // Logged here, where the backend was actually queried, rather than while rendering the response. + // A failure is cached (see [`CachedResponse`]), so logging it per response would produce a line + // for every request that hits the cached failure, which is precisely the burst we cache to avoid. + resource_info.inspect_err(|error| { + let source = error as &dyn std::error::Error; + + if http_error::Error::status_code(error).is_client_error() { + // The caller asked for something that cannot be looked up, such as a name no DataHub URN + // can express. That is their problem and says nothing about the health of this process or + // of the backend, so it does not belong in the log by default. Any user who can name a + // resource can produce these at will. + tracing::debug!(error = source, "Rejected a resource information request"); + } else { + tracing::warn!(error = source, "Failed to look up resource information"); + } + }) +} + +#[cfg(test)] +mod tests { + use wiremock::{ + Mock, MockServer, ResponseTemplate, + matchers::{method, path}, + }; + + use super::*; + use crate::api::RawIdentifier; + + /// State whose backend queries `mock_server`, with the cache a cluster gets by default. + fn state_for(mock_server: &MockServer) -> AppState { + let graphql_url = format!("{}/api/graphql", mock_server.uri()) + .parse() + .expect("the mock server's address must be a valid URL"); + + AppState { + backend: Arc::new(ResolvedBackend::DataHub( + backend::data_hub::ResolvedDataHubBackend::for_tests(graphql_url), + )), + resource_info_cache: cache::build(&Default::default()), + } + } + + fn request() -> ResourceInfoRequest { + ResourceInfoRequest::RawIdentifier(RawIdentifier { + identifier: "urn:li:dataset:(urn:li:dataPlatform:trino,broken,PROD)".into(), + }) + } - Ok(Json(resource_info)) + /// A lookup that keeps failing must not query the backend again on every request. Anyone who can + /// name a resource could otherwise amplify their requests against DataHub, and worst of all + /// exactly while DataHub is already unwell. + #[tokio::test] + async fn a_failing_lookup_queries_the_backend_only_once() { + let mock_server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/api/graphql")) + .respond_with(ResponseTemplate::new(500).set_body_string("GMS is having a bad day")) + .expect(1) + .mount(&mock_server) + .await; + + let state = state_for(&mock_server); + for _ in 0..5 { + let result = get_resource_info(state.clone(), request()).await; + assert!(result.is_err(), "the lookup must keep failing"); + } + + // The mock's `.expect(1)` is verified when `mock_server` is dropped. + } } diff --git a/rust/user-info-fetcher/src/backend/entra.rs b/rust/user-info-fetcher/src/backend/entra.rs index 3cc9dd76..daa9e025 100644 --- a/rust/user-info-fetcher/src/backend/entra.rs +++ b/rust/user-info-fetcher/src/backend/entra.rs @@ -1,7 +1,11 @@ -use std::{collections::HashMap, path::Path}; +use std::{collections::HashMap, path::Path, time::Duration}; use hyper::StatusCode; -use info_fetcher_commons::utils::{self, http::send_json_request}; +use info_fetcher_commons::utils::{ + self, + http::send_json_request, + token::{CachedToken, MintedToken}, +}; use serde::Deserialize; use snafu::{ResultExt, Snafu}; use stackable_opa_operator::crd::user_info_fetcher::v1alpha2; @@ -80,6 +84,10 @@ impl http_error::Error for Error { #[derive(Deserialize)] struct OAuthResponse { access_token: String, + + /// How many seconds the token stays valid, which lets us cache it rather than mint one per + /// lookup. Treated as optional so a response without it degrades to minting per lookup. + expires_in: Option, } #[derive(Clone, Deserialize)] @@ -123,6 +131,9 @@ pub struct ResolvedEntraBackend { client_id: String, client_secret: String, http_client: reqwest::Client, + + /// The OAuth2 access token, minted on demand and reused until it is about to expire. + access_token: CachedToken, } impl ResolvedEntraBackend { @@ -165,6 +176,7 @@ impl ResolvedEntraBackend { client_id, client_secret, http_client, + access_token: CachedToken::new(), }) } @@ -186,23 +198,67 @@ impl ResolvedEntraBackend { TlsClientDetails { tls: tls.clone() }.uses_tls(), )?; - let token_url = entra_backend.oauth2_token(); - let authn = send_json_request::(self.http_client.post(token_url).form(&[ - ("client_id", self.client_id.as_str()), - ("client_secret", self.client_secret.as_str()), - ("scope", "https://graph.microsoft.com/.default"), - ("grant_type", "client_credentials"), - ])) - .await - .context(AccessTokenSnafu)?; + let access_token = self.access_token(&entra_backend).await?; + match self + .get_user_info_with(req, &entra_backend, &access_token) + .await + { + Err(error) if utils::http::is_unauthorized(&error) => { + // The token was accepted when it was minted, so it has stopped being valid ahead of + // its stated expiry. It was revoked, or the issuer's and our clock disagree. Drop + // it and give the lookup exactly one more go with a fresh one. + tracing::warn!( + error = &error as &dyn std::error::Error, + "Entra rejected the cached access token; re-authenticating and retrying once" + ); + self.access_token.invalidate().await; + let access_token = self.access_token(&entra_backend).await?; + self.get_user_info_with(req, &entra_backend, &access_token) + .await + } + result => result, + } + } + + /// The cached access token, minting one if there is none or it is about to expire. + async fn access_token(&self, entra_backend: &EntraBackend) -> Result { + self.access_token + .get(|| async { + let response = send_json_request::( + self.http_client.post(entra_backend.oauth2_token()).form(&[ + ("client_id", self.client_id.as_str()), + ("client_secret", self.client_secret.as_str()), + ("scope", "https://graph.microsoft.com/.default"), + ("grant_type", "client_credentials"), + ]), + ) + .await + .context(AccessTokenSnafu)?; + + Ok(MintedToken { + token: response.access_token, + lifetime: response.expires_in.map(Duration::from_secs), + }) + }) + .await + } + + /// Looks the user up with an already-obtained `access_token`, so the caller can retry with a fresh + /// one if this token turns out to be rejected. + async fn get_user_info_with( + &self, + req: &UserInfoRequest, + entra_backend: &EntraBackend, + access_token: &str, + ) -> Result { let user_info = match req { UserInfoRequest::UserInfoRequestById(req) => { let user_id = &req.id; send_json_request::( self.http_client .get(entra_backend.user_info(user_id)) - .bearer_auth(&authn.access_token), + .bearer_auth(access_token), ) .await .with_context(|_| UserNotFoundByIdSnafu { @@ -214,7 +270,7 @@ impl ResolvedEntraBackend { send_json_request::( self.http_client .get(entra_backend.user_info(username)) - .bearer_auth(&authn.access_token), + .bearer_auth(access_token), ) .await .with_context(|_| SearchForUserSnafu { @@ -242,7 +298,7 @@ impl ResolvedEntraBackend { pages_remaining -= 1; let response = send_json_request::( - self.http_client.get(url).bearer_auth(&authn.access_token), + self.http_client.get(url).bearer_auth(access_token), ) .await .with_context(|_| RequestUserGroupsSnafu { @@ -373,10 +429,31 @@ mod tests { }, client_id: "client-id".to_owned(), client_secret: "client-secret".to_owned(), + access_token: CachedToken::new(), http_client: reqwest::Client::new(), } } + /// Mounts the OAuth2 token endpoint, answering with a token that is valid for `expires_in` + /// seconds. Without an `expires_in` the token is deliberately not cached, see + /// [`utils::token::CachedToken`]. + /// + /// `expected_calls` asserts how often the endpoint is hit; wiremock verifies it when the server + /// is dropped. + async fn mock_token(mock_server: &MockServer, expires_in: Option, expected_calls: u64) { + let mut body = serde_json::json!({"access_token": "access-token"}); + if let Some(expires_in) = expires_in { + body["expires_in"] = expires_in.into(); + } + + Mock::given(method("POST")) + .and(path(format!("/{TENANT_ID}/oauth2/v2.0/token"))) + .respond_with(ResponseTemplate::new(200).set_body_json(body)) + .expect(expected_calls) + .mount(mock_server) + .await; + } + /// Mocks the OAuth2 token and user metadata endpoints, which every `get_user_info` call hits /// before it gets to the group memberships we actually care about. async fn mock_token_and_user(mock_server: &MockServer) { @@ -482,6 +559,63 @@ mod tests { assert_eq!(user_info.groups.len(), MAX_GROUP_PAGES); } + /// Mounts the user metadata and (empty) group endpoints, so a lookup gets all the way through. + async fn mock_user_and_groups(mock_server: &MockServer, user_status: u16) { + Mock::given(method("GET")) + .and(path(format!("/v1.0/users/{USER_ID}"))) + .respond_with( + ResponseTemplate::new(user_status).set_body_json(serde_json::json!({ + "id": USER_ID, + "userPrincipalName": "alice@example.com", + })), + ) + .mount(mock_server) + .await; + + Mock::given(method("GET")) + .and(path(format!("/v1.0/users/{USER_ID}/memberOf"))) + .respond_with( + ResponseTemplate::new(200).set_body_json(serde_json::json!({"value": []})), + ) + .mount(mock_server) + .await; + } + + /// The access token was minted for every single lookup, doubling the round trips. Entra reports + /// how long it is valid for, so it only has to be minted once. + #[tokio::test] + async fn test_entra_reuses_a_cached_access_token_across_lookups() { + let mock_server = MockServer::start().await; + mock_token(&mock_server, Some(3600), 1).await; + mock_user_and_groups(&mock_server, 200).await; + + let backend = backend_for(&mock_server); + get_user_info_by_id(&backend).await; + get_user_info_by_id(&backend).await; + + // The token endpoint's `.expect(1)` is verified when `mock_server` is dropped. + } + + /// A token can stop being accepted before it expires, e.g. by being revoked. The rejection is the + /// only way to find that out, so it has to trigger exactly one re-authentication. Not none + /// (the lookup would keep failing until the token expired), and not a retry loop. + #[tokio::test] + async fn test_entra_reauthenticates_once_when_the_token_is_rejected() { + let mock_server = MockServer::start().await; + mock_token(&mock_server, Some(3600), 2).await; + mock_user_and_groups(&mock_server, 401).await; + + let error = backend_for(&mock_server) + .get_user_info(&UserInfoRequest::UserInfoRequestById(UserInfoRequestById { + id: USER_ID.to_owned(), + })) + .await + .expect_err("a permanently rejected token must surface as an error"); + + assert!(utils::http::is_unauthorized(&error), "{error}"); + // The token endpoint's `.expect(2)` is verified when `mock_server` is dropped. + } + #[test] fn test_entra_defaults_id() { let tenant_id = "1234-5678-1234-5678"; diff --git a/rust/user-info-fetcher/src/backend/keycloak.rs b/rust/user-info-fetcher/src/backend/keycloak.rs index e26cf9b1..7367f564 100644 --- a/rust/user-info-fetcher/src/backend/keycloak.rs +++ b/rust/user-info-fetcher/src/backend/keycloak.rs @@ -1,11 +1,16 @@ -use std::{collections::HashMap, path::Path}; +use std::{collections::HashMap, path::Path, time::Duration}; use hyper::StatusCode; -use info_fetcher_commons::utils::{self, http::send_json_request}; +use info_fetcher_commons::utils::{ + self, + http::send_json_request, + token::{CachedToken, MintedToken}, +}; use serde::Deserialize; use snafu::{OptionExt, ResultExt, Snafu}; use stackable_opa_operator::crd::user_info_fetcher::v1alpha2; use stackable_operator::crd::authentication::oidc; +use url::Url; use crate::{UserInfo, UserInfoRequest, http_error}; @@ -85,6 +90,10 @@ impl http_error::Error for Error { #[derive(Deserialize)] struct OAuthResponse { access_token: String, + + /// How many seconds the token stays valid, which lets us cache it rather than mint one per + /// lookup. Treated as optional so a response without it degrades to minting per lookup. + expires_in: Option, } /// The minimal structure of [UserRepresentation] that is returned by [`/users`][users] and [`/users/{id}`][user-by-id]. @@ -118,6 +127,9 @@ pub struct ResolvedKeycloakBackend { client_id: String, client_secret: String, http_client: reqwest::Client, + + /// The OAuth2 access token, minted on demand and reused until it is about to expire. + access_token: CachedToken, } impl ResolvedKeycloakBackend { @@ -155,18 +167,44 @@ impl ResolvedKeycloakBackend { client_id, client_secret, http_client, + access_token: CachedToken::new(), }) } pub(crate) async fn get_user_info(&self, req: &UserInfoRequest) -> Result { + let keycloak_url = self.keycloak_url()?; + + let access_token = self.access_token(&keycloak_url).await?; + match self + .get_user_info_with(req, &keycloak_url, &access_token) + .await + { + Err(error) if utils::http::is_unauthorized(&error) => { + // The token was accepted when it was minted, so it has stopped being valid ahead of + // its stated expiry. It was revoked, or the issuer's and our clock disagree. Drop + // it and give the lookup exactly one more go with a fresh one. + tracing::warn!( + error = &error as &dyn std::error::Error, + "Keycloak rejected the cached access token; re-authenticating and retrying once" + ); + self.access_token.invalidate().await; + + let access_token = self.access_token(&keycloak_url).await?; + self.get_user_info_with(req, &keycloak_url, &access_token) + .await + } + result => result, + } + } + + /// The base URL of the configured Keycloak. + fn keycloak_url(&self) -> Result { let v1alpha2::KeycloakBackend { - client_credentials_secret: _, - admin_realm, - user_realm, hostname, port, root_path, tls, + .. } = &self.config; // We re-use existent functionality from operator-rs, besides it being a bit of miss-use. @@ -180,24 +218,50 @@ impl ResolvedKeycloakBackend { Vec::new(), None, ); - let keycloak_url = wrapping_auth_provider + + wrapping_auth_provider .endpoint_url() - .context(ParseOidcEndpointUrlSnafu)?; + .context(ParseOidcEndpointUrlSnafu) + } - let authn = send_json_request::( - self.http_client - .post( - keycloak_url - .join(&format!( - "realms/{admin_realm}/protocol/openid-connect/token" - )) - .context(ConstructOidcEndpointPathSnafu)?, + /// The cached access token, minting one if there is none or it is about to expire. + async fn access_token(&self, keycloak_url: &Url) -> Result { + let admin_realm = &self.config.admin_realm; + + self.access_token + .get(|| async { + let response = send_json_request::( + self.http_client + .post( + keycloak_url + .join(&format!( + "realms/{admin_realm}/protocol/openid-connect/token" + )) + .context(ConstructOidcEndpointPathSnafu)?, + ) + .basic_auth(&self.client_id, Some(&self.client_secret)) + .form(&[("grant_type", "client_credentials")]), ) - .basic_auth(&self.client_id, Some(&self.client_secret)) - .form(&[("grant_type", "client_credentials")]), - ) - .await - .context(AccessTokenSnafu)?; + .await + .context(AccessTokenSnafu)?; + + Ok(MintedToken { + token: response.access_token, + lifetime: response.expires_in.map(Duration::from_secs), + }) + }) + .await + } + + /// Looks the user up with an already-obtained `access_token`, so the caller can retry with a fresh + /// one if this token turns out to be rejected. + async fn get_user_info_with( + &self, + req: &UserInfoRequest, + keycloak_url: &Url, + access_token: &str, + ) -> Result { + let user_realm = &self.config.user_realm; let users_base_url = keycloak_url .join(&format!("admin/realms/{user_realm}/users/")) @@ -213,7 +277,7 @@ impl ResolvedKeycloakBackend { .join(&req.id) .context(ConstructOidcEndpointPathSnafu)?, ) - .bearer_auth(&authn.access_token), + .bearer_auth(access_token), ) .await .context(UserNotFoundByIdSnafu { user_id })? @@ -225,9 +289,7 @@ impl ResolvedKeycloakBackend { .context(ConstructOidcEndpointPathSnafu)?; let users = send_json_request::>( - self.http_client - .get(users_url) - .bearer_auth(&authn.access_token), + self.http_client.get(users_url).bearer_auth(access_token), ) .await .context(SearchForUserSnafu)?; @@ -250,7 +312,7 @@ impl ResolvedKeycloakBackend { .join(&format!("{}/groups", user_info.id)) .context(ConstructOidcEndpointPathSnafu)?, ) - .bearer_auth(&authn.access_token), + .bearer_auth(access_token), ) .await .context(RequestUserGroupsSnafu { @@ -266,3 +328,126 @@ impl ResolvedKeycloakBackend { }) } } + +#[cfg(test)] +mod tests { + use std::str::FromStr; + + use stackable_operator::{ + commons::{networking::HostName, tls_verification::TlsClientDetails}, + v2::types::kubernetes::SecretName, + }; + use wiremock::{ + Mock, MockServer, ResponseTemplate, + matchers::{method, path}, + }; + + use super::*; + use crate::{UserInfoRequestById, backend::keycloak::ResolvedKeycloakBackend}; + + const ADMIN_REALM: &str = "master"; + const USER_REALM: &str = "my-realm"; + const USER_ID: &str = "8765-4321-8765-4321"; + + /// Builds a backend pointing at `mock_server`, bypassing [`ResolvedKeycloakBackend::resolve`] so + /// that no credentials have to be read from disk. + fn backend_for(mock_server: &MockServer) -> ResolvedKeycloakBackend { + ResolvedKeycloakBackend { + config: v1alpha2::KeycloakBackend { + hostname: HostName::from_str(&mock_server.address().ip().to_string()).unwrap(), + port: Some(mock_server.address().port()), + root_path: "/".to_owned(), + tls: TlsClientDetails { tls: None }, + client_credentials_secret: SecretName::from_str("keycloak-credentials").unwrap(), + admin_realm: ADMIN_REALM.to_owned(), + user_realm: USER_REALM.to_owned(), + }, + client_id: "client-id".to_owned(), + client_secret: "client-secret".to_owned(), + http_client: reqwest::Client::new(), + access_token: CachedToken::new(), + } + } + + /// Mounts the token endpoint, answering with a token valid for `expires_in` seconds. + /// `expected_calls` is verified by wiremock when the server is dropped. + async fn mock_token(mock_server: &MockServer, expires_in: Option, expected_calls: u64) { + let mut body = serde_json::json!({"access_token": "access-token"}); + if let Some(expires_in) = expires_in { + body["expires_in"] = expires_in.into(); + } + + Mock::given(method("POST")) + .and(path(format!( + "/realms/{ADMIN_REALM}/protocol/openid-connect/token" + ))) + .respond_with(ResponseTemplate::new(200).set_body_json(body)) + .expect(expected_calls) + .mount(mock_server) + .await; + } + + /// Mounts the user metadata and (empty) group endpoints, so a lookup gets all the way through. + async fn mock_user_and_groups(mock_server: &MockServer, user_status: u16) { + Mock::given(method("GET")) + .and(path(format!("/admin/realms/{USER_REALM}/users/{USER_ID}"))) + .respond_with( + ResponseTemplate::new(user_status).set_body_json(serde_json::json!({ + "id": USER_ID, + "username": "alice", + })), + ) + .mount(mock_server) + .await; + + Mock::given(method("GET")) + .and(path(format!( + "/admin/realms/{USER_REALM}/users/{USER_ID}/groups" + ))) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!([]))) + .mount(mock_server) + .await; + } + + async fn get_user_info_by_id( + backend: &ResolvedKeycloakBackend, + ) -> Result { + backend + .get_user_info(&UserInfoRequest::UserInfoRequestById(UserInfoRequestById { + id: USER_ID.to_owned(), + })) + .await + } + + /// The access token was minted for every single lookup, doubling the round trips. Keycloak reports + /// how long it is valid for, so it only has to be minted once. + #[tokio::test] + async fn keycloak_reuses_a_cached_access_token_across_lookups() { + let mock_server = MockServer::start().await; + mock_token(&mock_server, Some(3600), 1).await; + mock_user_and_groups(&mock_server, 200).await; + + let backend = backend_for(&mock_server); + get_user_info_by_id(&backend).await.expect("lookup works"); + get_user_info_by_id(&backend).await.expect("lookup works"); + + // The token endpoint's `.expect(1)` is verified when `mock_server` is dropped. + } + + /// A token can stop being accepted before it expires, e.g. by being revoked. The rejection is the + /// only way to find that out, so it has to trigger exactly one re-authentication. Not none + /// (the lookup would keep failing until the token expired), and not a retry loop. + #[tokio::test] + async fn keycloak_reauthenticates_once_when_the_token_is_rejected() { + let mock_server = MockServer::start().await; + mock_token(&mock_server, Some(3600), 2).await; + mock_user_and_groups(&mock_server, 401).await; + + let error = get_user_info_by_id(&backend_for(&mock_server)) + .await + .expect_err("a permanently rejected token must surface as an error"); + + assert!(utils::http::is_unauthorized(&error), "{error}"); + // The token endpoint's `.expect(2)` is verified when `mock_server` is dropped. + } +} diff --git a/tests/templates/kuttl/logging/03-install-opa.yaml.j2 b/tests/templates/kuttl/logging/03-install-opa.yaml.j2 index bf29f3ee..f6b7bef2 100644 --- a/tests/templates/kuttl/logging/03-install-opa.yaml.j2 +++ b/tests/templates/kuttl/logging/03-install-opa.yaml.j2 @@ -17,6 +17,16 @@ data: false } --- +apiVersion: v1 +kind: Secret +metadata: + name: datahub-credentials +stringData: + # The resource-info-fetcher only reads this token at startup; this test never sends it a request, + # so the token does not have to be valid and the configured DataHub does not have to exist. The + # sidecar is only here so that its logs can be asserted on. + token: not-a-real-datahub-token +--- apiVersion: opa.stackable.tech/v1alpha1 kind: OpaCluster metadata: @@ -32,6 +42,16 @@ spec: pullPolicy: IfNotPresent clusterConfig: vectorAggregatorConfigMapName: opa-vector-aggregator-discovery + # Both info-fetcher sidecars are enabled so that the test covers the collection of their file + # logs as well. Neither of them is queried by this test. + userInfo: + backend: + none: {} + resourceInfo: + backend: + dataHub: + hostname: datahub-gms + credentialsSecretName: datahub-credentials servers: roleGroups: automatic-log-config: @@ -52,6 +72,16 @@ spec: level: NONE file: level: INFO + user-info-fetcher: + console: + level: NONE + file: + level: INFO + resource-info-fetcher: + console: + level: NONE + file: + level: INFO vector: console: level: INFO diff --git a/tests/templates/kuttl/logging/opa-vector-aggregator-values.yaml.j2 b/tests/templates/kuttl/logging/opa-vector-aggregator-values.yaml.j2 index 4bb9fdc4..64f3e288 100644 --- a/tests/templates/kuttl/logging/opa-vector-aggregator-values.yaml.j2 +++ b/tests/templates/kuttl/logging/opa-vector-aggregator-values.yaml.j2 @@ -44,6 +44,18 @@ customConfig: condition: >- starts_with(string!(.pod), "test-opa-server-automatic-log-config") && .container == "bundle-builder" + filteredAutomaticLogConfigServerUserInfoFetcher: + type: filter + inputs: [validEvents] + condition: >- + starts_with(string!(.pod), "test-opa-server-automatic-log-config") && + .container == "user-info-fetcher" + filteredAutomaticLogConfigServerResourceInfoFetcher: + type: filter + inputs: [validEvents] + condition: >- + starts_with(string!(.pod), "test-opa-server-automatic-log-config") && + .container == "resource-info-fetcher" filteredAutomaticLogConfigServerVector: type: filter inputs: [validEvents]