Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -45,8 +45,7 @@ impl FunctionRunnerService for Function {
// Copies the request's tag, desired state, and context forward.
let mut rsp = response::to(&req, response::DEFAULT_TTL);

let desired = rsp.desired.get_or_insert_default();
let bucket = desired.resources.entry("bucket".to_string()).or_default();
let bucket = response::desired_composed(&mut rsp, "bucket");
resource::update(bucket, &serde_json::json!({
"apiVersion": "s3.aws.upbound.io/v1beta2",
"kind": "Bucket",
Expand Down
9 changes: 3 additions & 6 deletions example/src/function.rs
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@

use function_sdk_rust::proto::v1::function_runner_service_server::FunctionRunnerService;
use function_sdk_rust::proto::v1::{RunFunctionRequest, RunFunctionResponse};
use function_sdk_rust::{resource, response};
use function_sdk_rust::{request, resource, response};
use tonic::{Request, Response, Status};

#[derive(Debug, Default)]
Expand Down Expand Up @@ -44,17 +44,14 @@ impl FunctionRunnerService for Function {
return Ok(Response::new(rsp));
};

let composite = req.observed.as_ref().and_then(|s| s.composite.as_ref());
let Ok(observed_xr) = resource::get::<Xr>(composite) else {
let Ok(observed_xr) = resource::get::<Xr>(request::observed_composite(&req)) else {
response::fatal(&mut rsp, "invalid XR: spec.region is required");
return Ok(Response::new(rsp));
};
let region = observed_xr.spec.region;

let desired = rsp.desired.get_or_insert_default();
let bucket = desired.resources.entry("bucket".to_string()).or_default();
resource::update(
bucket,
response::desired_composed(&mut rsp, "bucket"),
&serde_json::json!({
"apiVersion": format!("s3.aws.upbound.io/{version}"),
"kind": "Bucket",
Expand Down
48 changes: 47 additions & 1 deletion sdk/src/request.rs
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
//! Helpers for working with RunFunctionRequests.

use crate::Error;
use crate::proto::v1::{Capability, CredentialData, RunFunctionRequest, credentials};
use crate::proto::v1::{Capability, CredentialData, Resource, RunFunctionRequest, credentials};
use crate::resource::{pb_to_json, struct_to_json};

/// The requirement name a WatchOperation uses to inject the watched resource.
Expand All @@ -19,6 +19,22 @@ pub fn get_input<T: serde::de::DeserializeOwned>(req: &RunFunctionRequest) -> Re
Ok(serde_json::from_value(struct_to_json(input))?)
}

/// Gets the observed composite resource (XR), if the request carries one.
///
/// Read it as a typed value with [`crate::resource::get`].
pub fn observed_composite(req: &RunFunctionRequest) -> Option<&Resource> {
req.observed.as_ref()?.composite.as_ref()
}

/// Gets an observed composed resource by name.
///
/// Returns None when Crossplane has not observed the resource, for example
/// before it is first created. Read it as a typed value with
/// [`crate::resource::get`].
pub fn observed_composed<'a>(req: &'a RunFunctionRequest, name: &str) -> Option<&'a Resource> {
req.observed.as_ref()?.resources.get(name)
}

/// Gets a function context value by key, if present.
///
/// See [`crate::context`] for well-known context keys.
Expand Down Expand Up @@ -55,6 +71,36 @@ pub fn get_required_resource(req: &RunFunctionRequest, name: &str) -> Option<ser
get_required_resources(req, name)?.into_iter().next()
}

/// Gets required resources by requirement name as typed values.
///
/// Returns `Ok(None)` when Crossplane has not (yet) resolved the
/// requirement, and an empty Vec when it resolved the requirement but found
/// no matches. Integral numbers are restored to integers first, as in
/// [`crate::resource::get`], so integer fields of generated models
/// deserialize.
pub fn get_required_resources_as<T: serde::de::DeserializeOwned>(
req: &RunFunctionRequest,
name: &str,
) -> Result<Option<Vec<T>>, Error> {
get_required_resources(req, name)
.map(|resources| resources.into_iter().map(serde_json::from_value).collect())
.transpose()
.map_err(Error::from)
}

/// Gets a single required resource by requirement name as a typed value.
///
/// Returns `Ok(None)` when the requirement is unresolved or matched nothing.
pub fn get_required_resource_as<T: serde::de::DeserializeOwned>(
req: &RunFunctionRequest,
name: &str,
) -> Result<Option<T>, Error> {
get_required_resource(req, name)
.map(serde_json::from_value)
.transpose()
.map_err(Error::from)
}

/// Gets the watched resource that triggered this operation, if any.
///
/// When a WatchOperation creates an Operation it injects the resource that
Expand Down
14 changes: 8 additions & 6 deletions sdk/src/resource.rs
Original file line number Diff line number Diff line change
Expand Up @@ -26,12 +26,14 @@ pub fn struct_to_json(s: &Struct) -> serde_json::Value {
/// Deserializes a resource's JSON representation into a typed value.
///
/// Works for both the composite resource (XR) and composed resources - both
/// are represented by [`Resource`]. Pass `req.observed.composite.as_ref()`
/// for the observed XR, or `req.observed.resources.get(name)` for an
/// observed composed resource; the desired equivalents (`rsp.desired...`)
/// work the same way. Returns [`Error::MissingResource`] when `resource` is
/// `None` or has no JSON representation set, for example a composed
/// resource Crossplane has not yet observed.
/// are represented by [`Resource`]. Pass
/// [`request::observed_composite`](crate::request::observed_composite) for
/// the observed XR, or
/// [`request::observed_composed`](crate::request::observed_composed) for an
/// observed composed resource; desired resources work the same way. Returns
/// [`Error::MissingResource`] when `resource` is `None` or has no JSON
/// representation set, for example a composed resource Crossplane has not
/// yet observed.
pub fn get<T: serde::de::DeserializeOwned>(resource: Option<&Resource>) -> Result<T, Error> {
let s = resource
.and_then(|r| r.resource.as_ref())
Expand Down
31 changes: 29 additions & 2 deletions sdk/src/response.rs
Original file line number Diff line number Diff line change
Expand Up @@ -7,8 +7,9 @@ use serde::Serialize;

use crate::Error;
use crate::proto::v1::{
Condition, MatchLabels, Requirements, ResourceSelector, Result as FnResult, RunFunctionRequest,
RunFunctionResponse, SchemaSelector, Severity, Status, Target, resource_selector,
Condition, MatchLabels, Requirements, Resource, ResourceSelector, Result as FnResult,
RunFunctionRequest, RunFunctionResponse, SchemaSelector, Severity, Status, Target,
resource_selector,
};
use crate::resource::{json_to_pb, json_to_struct};

Expand Down Expand Up @@ -36,6 +37,32 @@ pub fn to(req: &RunFunctionRequest, ttl: Duration) -> RunFunctionResponse {
}
}

/// Gets the desired composite resource (XR), inserting an empty one when
/// the response has none.
///
/// A function may only set the desired XR's status; Crossplane ignores its
/// desired metadata and spec. Write it with [`crate::resource::update_status`].
pub fn desired_composite(rsp: &mut RunFunctionResponse) -> &mut Resource {
rsp.desired
.get_or_insert_default()
.composite
.get_or_insert_default()
}

/// Gets a desired composed resource by name, inserting an empty one when
/// the response has none.
///
/// The resource is modified in place, so whatever earlier pipeline steps
/// desired for it, copied forward by [`to`], is kept unless overwritten.
/// Write it with [`crate::resource::update`].
pub fn desired_composed(rsp: &mut RunFunctionResponse, name: impl Into<String>) -> &mut Resource {
rsp.desired
.get_or_insert_default()
.resources
.entry(name.into())
.or_default()
}

/// Adds a normal result to the response.
///
/// Returns the added result so a reason or target can be set on it:
Expand Down
174 changes: 174 additions & 0 deletions sdk/tests/helpers.rs
Original file line number Diff line number Diff line change
Expand Up @@ -231,6 +231,180 @@ fn required_resources_distinguish_unresolved_from_empty() {
assert_eq!(request::get_required_resources(&req, "empty"), Some(vec![]));
}

#[test]
fn typed_required_resources_deserialize_items() {
#[derive(serde::Deserialize, Debug, PartialEq)]
struct Metadata {
name: String,
}
#[derive(serde::Deserialize, Debug, PartialEq)]
struct Spec {
replicas: i64,
}
#[derive(serde::Deserialize, Debug, PartialEq)]
struct Provider {
metadata: Metadata,
spec: Spec,
}

// Integers arrive as f64 through structpb, like observed resources.
let req: RunFunctionRequest = serde_json::from_str(
r#"{"requiredResources": {
"providers": {"items": [
{"resource": {"metadata": {"name": "aws"}, "spec": {"replicas": 3.0}}},
{"resource": {"metadata": {"name": "gcp"}, "spec": {"replicas": 1.0}}}
]}
}}"#,
)
.unwrap();

let providers: Vec<Provider> = request::get_required_resources_as(&req, "providers")
.unwrap()
.unwrap();
assert_eq!(
providers,
vec![
Provider {
metadata: Metadata {
name: "aws".to_string()
},
spec: Spec { replicas: 3 },
},
Provider {
metadata: Metadata {
name: "gcp".to_string()
},
spec: Spec { replicas: 1 },
},
]
);
let first: Provider = request::get_required_resource_as(&req, "providers")
.unwrap()
.unwrap();
assert_eq!(first.metadata.name, "aws");

let untyped = request::get_required_resources_as::<serde_json::Value>(&req, "providers");
assert_eq!(
untyped.unwrap(),
request::get_required_resources(&req, "providers")
);
}

#[test]
fn typed_required_resources_distinguish_unresolved_from_empty() {
let req: RunFunctionRequest =
serde_json::from_str(r#"{"requiredResources": {"empty": {}}}"#).unwrap();

let unresolved = request::get_required_resources_as::<serde_json::Value>(&req, "nope");
assert_eq!(unresolved.unwrap(), None);
let empty = request::get_required_resources_as::<serde_json::Value>(&req, "empty");
assert_eq!(empty.unwrap(), Some(vec![]));

let unresolved = request::get_required_resource_as::<serde_json::Value>(&req, "nope");
assert_eq!(unresolved.unwrap(), None);
let empty = request::get_required_resource_as::<serde_json::Value>(&req, "empty");
assert_eq!(empty.unwrap(), None);
}

#[test]
fn typed_required_resources_report_mismatched_items() {
#[derive(serde::Deserialize, Debug)]
#[expect(dead_code, reason = "the item never deserializes")]
struct Item {
replicas: i64,
}

let req: RunFunctionRequest = serde_json::from_str(
r#"{"requiredResources": {"providers": {"items": [
{"resource": {"replicas": "three"}}
]}}}"#,
)
.unwrap();

let all = request::get_required_resources_as::<Item>(&req, "providers");
assert!(matches!(all, Err(Error::Json(_))));
let one = request::get_required_resource_as::<Item>(&req, "providers");
assert!(matches!(one, Err(Error::Json(_))));
}

#[test]
fn observed_accessors_borrow_the_observed_state() {
let req: RunFunctionRequest = serde_json::from_str(
r#"{"observed": {
"composite": {"resource": {"kind": "XR"}},
"resources": {"bucket": {"resource": {"kind": "Bucket"}}}
}}"#,
)
.unwrap();
let observed = req.observed.as_ref().unwrap();

assert_eq!(
request::observed_composite(&req),
observed.composite.as_ref()
);
assert_eq!(
request::observed_composed(&req, "bucket"),
observed.resources.get("bucket")
);
assert_eq!(request::observed_composed(&req, "nope"), None);

let empty = RunFunctionRequest::default();
assert_eq!(request::observed_composite(&empty), None);
assert_eq!(request::observed_composed(&empty, "bucket"), None);
}

#[test]
fn desired_accessors_insert_missing_resources() {
let mut rsp = RunFunctionResponse::default();

resource::update_status(
response::desired_composite(&mut rsp),
&serde_json::json!({"bucket": "ready"}),
)
.unwrap();
resource::set_ready(response::desired_composite(&mut rsp), Ready::True);
resource::update(
response::desired_composed(&mut rsp, "bucket"),
&serde_json::json!({"kind": "Bucket"}),
)
.unwrap();

let desired = rsp.desired.unwrap();
let xr = desired.composite.as_ref().unwrap();
assert_eq!(xr.ready, Ready::True as i32);
let xr: serde_json::Value = resource::get(Some(xr)).unwrap();
assert_eq!(xr, serde_json::json!({"status": {"bucket": "ready"}}));
let bucket: serde_json::Value = resource::get(desired.resources.get("bucket")).unwrap();
assert_eq!(bucket, serde_json::json!({"kind": "Bucket"}));
}

#[test]
fn desired_accessors_keep_the_copied_forward_state() {
let mut rsp = response::to(&fixture(), response::DEFAULT_TTL);

let bucket = response::desired_composed(&mut rsp, "bucket");
resource::set_ready(bucket, Ready::True);
resource::update(
bucket,
&serde_json::json!({"spec": {"forProvider": {"region": "eu-central-1"}}}),
)
.unwrap();

let desired = rsp.desired.unwrap();
assert_eq!(desired.resources.len(), 1);
let bucket = &desired.resources["bucket"];
assert_eq!(bucket.ready, Ready::True as i32);
let bucket: serde_json::Value = resource::get(Some(bucket)).unwrap();
assert_eq!(
bucket,
serde_json::json!({
"apiVersion": "s3.aws.upbound.io/v1beta2",
"kind": "Bucket",
"spec": {"forProvider": {"region": "eu-central-1"}},
})
);
}

#[test]
fn require_resources_builds_selectors() {
let mut rsp = RunFunctionResponse::default();
Expand Down
Loading