mirror of
https://github.com/openobserve/openobserve.git
synced 2026-11-06 00:09:22 +00:00
feat(cloud): support contract subscription (#11336)
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Co-authored-by: claude[bot] <41898282+claude[bot]@users.noreply.github.com> Co-authored-by: udev <undefined@users.noreply.github.com>
This commit is contained in:
co-authored by
Claude Opus 4.7
claude[bot]
udev
parent
1033ade8bc
commit
b7bc228e49
@@ -39,16 +39,21 @@ pub struct ListSubscriptionResponseBody {
|
||||
pub subscription_type: String,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub customer_id: Option<String>,
|
||||
/// Billing provider: "stripe" or "aws"
|
||||
pub provider: String,
|
||||
/// Billing provider: "stripe", "aws", or "azure". Omitted for external-contract subs.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub provider: Option<String>,
|
||||
}
|
||||
|
||||
impl From<cloud_billings::CustomerBilling> for ListSubscriptionResponseBody {
|
||||
fn from(value: cloud_billings::CustomerBilling) -> Self {
|
||||
let provider = match value.provider {
|
||||
cloud_billings::MeteringProvider::NoOp => None,
|
||||
other => Some(other.to_string()),
|
||||
};
|
||||
Self {
|
||||
subscription_type: value.subscription_type.to_string(),
|
||||
customer_id: value.customer_id,
|
||||
provider: value.provider.to_string(),
|
||||
provider,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -101,3 +106,165 @@ pub struct NewUserAttribution {
|
||||
pub from: String,
|
||||
pub company: String,
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use cloud_billings::{
|
||||
CustomerBilling, MeteringProvider, SubscriptionType, org_usage::OrgUsageQueryResult,
|
||||
};
|
||||
|
||||
use super::*;
|
||||
|
||||
fn billing_with_provider(provider: MeteringProvider) -> CustomerBilling {
|
||||
let mut cb = CustomerBilling::new("admin@example.com", "org-x");
|
||||
cb.provider = provider;
|
||||
cb.subscription_type = SubscriptionType::Rate;
|
||||
cb.customer_id = Some("cus_abc".to_string());
|
||||
cb
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_list_subscription_response_omits_provider_for_noop() {
|
||||
// External-contract orgs (NoOp provider) must hide the provider field
|
||||
// from the API response — that's the contract the frontend relies on
|
||||
// to detect "this is an external contract" vs. a real billing backend.
|
||||
let mut cb = billing_with_provider(MeteringProvider::NoOp);
|
||||
cb.subscription_type = SubscriptionType::ExternalContract;
|
||||
cb.customer_id = Some("custom".to_string());
|
||||
|
||||
let body: ListSubscriptionResponseBody = cb.into();
|
||||
|
||||
assert!(body.provider.is_none());
|
||||
assert_eq!(body.subscription_type, "external-contract");
|
||||
assert_eq!(body.customer_id.as_deref(), Some("custom"));
|
||||
|
||||
let json = serde_json::to_value(&body).unwrap();
|
||||
assert!(
|
||||
json.get("provider").is_none(),
|
||||
"provider field must be omitted from JSON for NoOp, got {json}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_list_subscription_response_includes_provider_for_stripe() {
|
||||
let body: ListSubscriptionResponseBody =
|
||||
billing_with_provider(MeteringProvider::Stripe).into();
|
||||
|
||||
assert_eq!(body.provider.as_deref(), Some("stripe"));
|
||||
assert_eq!(body.subscription_type, "pay-as-you-go");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_list_subscription_response_includes_provider_for_aws_and_azure() {
|
||||
let aws: ListSubscriptionResponseBody = billing_with_provider(MeteringProvider::Aws).into();
|
||||
assert_eq!(aws.provider.as_deref(), Some("aws"));
|
||||
|
||||
let azure: ListSubscriptionResponseBody =
|
||||
billing_with_provider(MeteringProvider::Azure).into();
|
||||
assert_eq!(azure.provider.as_deref(), Some("azure"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_convert_to_unit_gb_to_mb() {
|
||||
let mut body = GetOrgUsageResponseBody {
|
||||
data: vec![OrgUserData {
|
||||
event: "ingestion".to_string(),
|
||||
value: 2.0,
|
||||
unit: "GB".to_string(),
|
||||
}],
|
||||
range: "30d".to_string(),
|
||||
};
|
||||
|
||||
body.convert_to_unit("MB");
|
||||
|
||||
assert_eq!(body.data[0].unit, "MB");
|
||||
assert!((body.data[0].value - 2048.0).abs() < f64::EPSILON);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_convert_to_unit_mb_to_gb() {
|
||||
let mut body = GetOrgUsageResponseBody {
|
||||
data: vec![OrgUserData {
|
||||
event: "ingestion".to_string(),
|
||||
value: 1024.0,
|
||||
unit: "MB".to_string(),
|
||||
}],
|
||||
range: "30d".to_string(),
|
||||
};
|
||||
|
||||
body.convert_to_unit("GB");
|
||||
|
||||
assert_eq!(body.data[0].unit, "GB");
|
||||
assert!((body.data[0].value - 1.0).abs() < f64::EPSILON);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_convert_to_unit_is_case_insensitive() {
|
||||
let mut body = GetOrgUsageResponseBody {
|
||||
data: vec![OrgUserData {
|
||||
event: "ingestion".to_string(),
|
||||
value: 1.0,
|
||||
unit: "gb".to_string(),
|
||||
}],
|
||||
range: "30d".to_string(),
|
||||
};
|
||||
|
||||
body.convert_to_unit("mb");
|
||||
|
||||
assert_eq!(body.data[0].unit, "MB");
|
||||
assert!((body.data[0].value - 1024.0).abs() < f64::EPSILON);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_convert_to_unit_no_op_when_units_match() {
|
||||
let mut body = GetOrgUsageResponseBody {
|
||||
data: vec![OrgUserData {
|
||||
event: "ingestion".to_string(),
|
||||
value: 5.5,
|
||||
unit: "GB".to_string(),
|
||||
}],
|
||||
range: "7d".to_string(),
|
||||
};
|
||||
|
||||
body.convert_to_unit("GB");
|
||||
|
||||
assert_eq!(body.data[0].unit, "GB");
|
||||
assert!((body.data[0].value - 5.5).abs() < f64::EPSILON);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_convert_to_unit_unknown_pair_is_ignored() {
|
||||
let mut body = GetOrgUsageResponseBody {
|
||||
data: vec![OrgUserData {
|
||||
event: "ingestion".to_string(),
|
||||
value: 100.0,
|
||||
unit: "Bytes".to_string(),
|
||||
}],
|
||||
range: "1d".to_string(),
|
||||
};
|
||||
|
||||
body.convert_to_unit("MB");
|
||||
|
||||
// No conversion path defined for Bytes → MB; record stays untouched.
|
||||
assert_eq!(body.data[0].unit, "Bytes");
|
||||
assert!((body.data[0].value - 100.0).abs() < f64::EPSILON);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_org_user_data_from_query_result() {
|
||||
// OrgUsageQueryResult only derives Deserialize, so build it from JSON.
|
||||
// UsageEvent uses default (PascalCase) serde, not snake_case.
|
||||
let json = serde_json::json!({
|
||||
"event": "Ingestion",
|
||||
"unit": "GB",
|
||||
"size": 12.5,
|
||||
});
|
||||
let q: OrgUsageQueryResult = serde_json::from_value(json).unwrap();
|
||||
|
||||
let data: OrgUserData = q.into();
|
||||
|
||||
assert!((data.value - 12.5).abs() < f64::EPSILON);
|
||||
assert_eq!(data.unit, "GB");
|
||||
assert_eq!(data.event, "Ingestion");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -27,12 +27,16 @@ use o2_enterprise::enterprise::common::config::get_config as get_o2_config;
|
||||
use {
|
||||
crate::common::meta::organization::OrganizationInvites,
|
||||
crate::common::meta::organization::{
|
||||
AllOrgListDetails, AllOrganizationResponse, ExtendTrialPeriodRequest,
|
||||
OrganizationInviteUserRecord,
|
||||
AllOrgListDetails, AllOrganizationResponse, CreateExternalContractRequest,
|
||||
ExtendExternalContractRequest, ExtendTrialPeriodRequest, OrganizationInviteUserRecord,
|
||||
},
|
||||
axum::body::Body,
|
||||
axum::http::StatusCode,
|
||||
o2_enterprise::enterprise::cloud::list_customer_billings,
|
||||
o2_enterprise::enterprise::cloud::{
|
||||
billings::{MeteringProvider, SubscriptionType},
|
||||
list_customer_billings,
|
||||
},
|
||||
svix_ksuid::KsuidLike,
|
||||
};
|
||||
|
||||
use crate::{
|
||||
@@ -188,30 +192,48 @@ pub async fn all_organizations(
|
||||
}
|
||||
};
|
||||
|
||||
let all_subscriptions = match list_customer_billings().await {
|
||||
Ok(orgs) => orgs
|
||||
.into_iter()
|
||||
.map(|cb| (cb.org_id, cb.subscription_type as i32))
|
||||
.collect::<HashMap<_, _>>(),
|
||||
let all_billings = match list_customer_billings().await {
|
||||
Ok(cbs) => cbs,
|
||||
Err(e) => {
|
||||
return MetaHttpResponse::internal_error(e.to_string());
|
||||
}
|
||||
};
|
||||
|
||||
let all_billing_info: HashMap<_, _> = all_billings
|
||||
.iter()
|
||||
.map(|cb| {
|
||||
(
|
||||
cb.org_id.clone(),
|
||||
(cb.subscription_type, cb.end_date, cb.provider.to_string()),
|
||||
)
|
||||
})
|
||||
.collect();
|
||||
|
||||
let mut id = 1;
|
||||
for org in all_orgs {
|
||||
let billing_info = all_billing_info.get(&org.identifier);
|
||||
let contract_end_date = billing_info.and_then(|(subscription, end_date, _)| {
|
||||
if *subscription == SubscriptionType::ExternalContract {
|
||||
*end_date
|
||||
} else {
|
||||
None
|
||||
}
|
||||
});
|
||||
let org = AllOrgListDetails {
|
||||
id,
|
||||
identifier: org.identifier.clone(),
|
||||
name: org.org_name,
|
||||
org_type: org.org_type.to_string(),
|
||||
plan: all_subscriptions
|
||||
.get(&org.identifier)
|
||||
.cloned()
|
||||
plan: billing_info
|
||||
.map(|(st, ..)| i16::from(*st) as i32)
|
||||
.unwrap_or_default(),
|
||||
created_at: org.created_at,
|
||||
updated_at: org.updated_at,
|
||||
trial_expires_at: Some(org.trial_ends_at),
|
||||
contract_end_date,
|
||||
billing_provider: billing_info
|
||||
.map(|(_, _, provider)| provider.clone())
|
||||
.unwrap_or_default(),
|
||||
};
|
||||
if !org_names.contains(&org.identifier) {
|
||||
org_names.insert(org.identifier.clone());
|
||||
@@ -560,6 +582,212 @@ pub async fn extend_trial_period(
|
||||
ret
|
||||
}
|
||||
|
||||
/// CreateExternalContract
|
||||
#[cfg(feature = "cloud")]
|
||||
#[utoipa::path(
|
||||
post,
|
||||
path = "/{org_id}/external_contract",
|
||||
context_path = "/api",
|
||||
tag = "Organizations",
|
||||
operation_id = "CreateExternalContract",
|
||||
summary = "Create an external contract subscription for an organization",
|
||||
security(
|
||||
("Authorization"= [])
|
||||
),
|
||||
request_body(content = inline(CreateExternalContractRequest), description = "Create external contract request", content_type = "application/json"),
|
||||
responses(
|
||||
(status = 200, description = "Success", content_type = "application/json", body = String),
|
||||
),
|
||||
extensions(
|
||||
("x-o2-mcp" = json!({"enabled": false}))
|
||||
)
|
||||
)]
|
||||
pub async fn create_external_contract(
|
||||
Path(org_id): Path<String>,
|
||||
Json(req): Json<CreateExternalContractRequest>,
|
||||
) -> Response {
|
||||
use o2_enterprise::enterprise::cloud::billings::CustomerBilling;
|
||||
|
||||
if org_id != "_meta" {
|
||||
return MetaHttpResponse::unauthorized("not authorized to access this resource");
|
||||
}
|
||||
|
||||
// Validate the target org exists
|
||||
if let Err(e) = infra::table::organizations::get(&req.org_id).await {
|
||||
return MetaHttpResponse::not_found(format!("Organization not found: {e}"));
|
||||
}
|
||||
|
||||
// Get the org admin's email for the billing record
|
||||
let admin = match infra::table::org_users::get_admin(&req.org_id).await {
|
||||
Ok(admin) => admin,
|
||||
Err(e) => {
|
||||
return MetaHttpResponse::not_found(format!(
|
||||
"Could not find admin for organization: {e}"
|
||||
));
|
||||
}
|
||||
};
|
||||
|
||||
let now = chrono::Utc::now().timestamp_micros();
|
||||
if req.end_date <= now {
|
||||
return MetaHttpResponse::bad_request("end_date must be in the future");
|
||||
}
|
||||
|
||||
// Block if the org already has any active billing record to avoid silently
|
||||
// overwriting a live Stripe/AWS/Azure subscription.
|
||||
if let Ok(Some(existing)) =
|
||||
o2_enterprise::enterprise::cloud::billings::get_billing_by_org_id(&req.org_id).await
|
||||
{
|
||||
let msg = if matches!(existing.provider, MeteringProvider::NoOp) {
|
||||
"Organization already has an active external contract"
|
||||
} else {
|
||||
log::warn!(
|
||||
"[EXT_CONTRACT] Refusing to create external contract for org {}: existing {} subscription would be overwritten",
|
||||
req.org_id,
|
||||
existing.provider
|
||||
);
|
||||
"Organization already has an active subscription; cancel it before creating an external contract"
|
||||
};
|
||||
return MetaHttpResponse::bad_request(msg);
|
||||
}
|
||||
|
||||
let mut billing = CustomerBilling::new(&admin.email, &req.org_id);
|
||||
billing.provider = MeteringProvider::NoOp;
|
||||
billing.subscription_type = SubscriptionType::ExternalContract;
|
||||
billing.customer_id = Some("custom".to_string());
|
||||
billing.subscription_id = Some(svix_ksuid::Ksuid::new(None, None).to_string());
|
||||
billing.end_date = Some(req.end_date);
|
||||
billing.expiry_notified_stage = None;
|
||||
|
||||
match o2_enterprise::enterprise::cloud::update_customer_billing(billing).await {
|
||||
Ok(_) => MetaHttpResponse::json(serde_json::json!({"status": "ok"})),
|
||||
Err(e) => {
|
||||
MetaHttpResponse::internal_error(format!("Failed to create external contract: {e}"))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// ExtendExternalContract
|
||||
#[cfg(feature = "cloud")]
|
||||
#[utoipa::path(
|
||||
put,
|
||||
path = "/{org_id}/external_contract",
|
||||
context_path = "/api",
|
||||
tag = "Organizations",
|
||||
operation_id = "ExtendExternalContract",
|
||||
summary = "Extend an external contract end date",
|
||||
security(
|
||||
("Authorization"= [])
|
||||
),
|
||||
request_body(content = inline(ExtendExternalContractRequest), description = "Extend external contract request", content_type = "application/json"),
|
||||
responses(
|
||||
(status = 200, description = "Success", content_type = "application/json", body = String),
|
||||
),
|
||||
extensions(
|
||||
("x-o2-mcp" = json!({"enabled": false}))
|
||||
)
|
||||
)]
|
||||
pub async fn extend_external_contract(
|
||||
Path(org_id): Path<String>,
|
||||
Json(req): Json<ExtendExternalContractRequest>,
|
||||
) -> Response {
|
||||
if org_id != "_meta" {
|
||||
return MetaHttpResponse::unauthorized("not authorized to access this resource");
|
||||
}
|
||||
|
||||
let mut billing = match o2_enterprise::enterprise::cloud::billings::get_billing_by_org_id(
|
||||
&req.org_id,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(Some(b)) if matches!(b.provider, MeteringProvider::NoOp) => b,
|
||||
Ok(_) => {
|
||||
return MetaHttpResponse::not_found(
|
||||
"No active external contract found for this organization",
|
||||
);
|
||||
}
|
||||
Err(e) => {
|
||||
return MetaHttpResponse::internal_error(e.to_string());
|
||||
}
|
||||
};
|
||||
|
||||
let now = chrono::Utc::now().timestamp_micros();
|
||||
if req.new_end_date <= now {
|
||||
return MetaHttpResponse::bad_request("new_end_date must be in the future");
|
||||
}
|
||||
if billing
|
||||
.end_date
|
||||
.is_some_and(|current| req.new_end_date <= current)
|
||||
{
|
||||
return MetaHttpResponse::bad_request(
|
||||
"new_end_date must be after the current contract end date",
|
||||
);
|
||||
}
|
||||
|
||||
billing.end_date = Some(req.new_end_date);
|
||||
// Reset expiry notifications so warnings fire again for the new window.
|
||||
billing.expiry_notified_stage = None;
|
||||
|
||||
match o2_enterprise::enterprise::cloud::update_customer_billing(billing).await {
|
||||
Ok(_) => MetaHttpResponse::json(serde_json::json!({"status": "ok"})),
|
||||
Err(e) => {
|
||||
MetaHttpResponse::internal_error(format!("Failed to extend external contract: {e}"))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// RevokeExternalContract
|
||||
#[cfg(feature = "cloud")]
|
||||
#[utoipa::path(
|
||||
delete,
|
||||
path = "/{org_id}/external_contract/{target_org_id}",
|
||||
context_path = "/api",
|
||||
tag = "Organizations",
|
||||
operation_id = "RevokeExternalContract",
|
||||
summary = "Revoke an external contract",
|
||||
security(
|
||||
("Authorization"= [])
|
||||
),
|
||||
params(
|
||||
("org_id" = String, Path, description = "Meta organization id"),
|
||||
("target_org_id" = String, Path, description = "Target organization id"),
|
||||
),
|
||||
responses(
|
||||
(status = 200, description = "Success", content_type = "application/json", body = String),
|
||||
),
|
||||
extensions(
|
||||
("x-o2-mcp" = json!({"enabled": false}))
|
||||
)
|
||||
)]
|
||||
pub async fn revoke_external_contract(
|
||||
Path((org_id, target_org_id)): Path<(String, String)>,
|
||||
) -> Response {
|
||||
if org_id != "_meta" {
|
||||
return MetaHttpResponse::unauthorized("not authorized to access this resource");
|
||||
}
|
||||
|
||||
// Verify the org has an external contract before deleting
|
||||
match o2_enterprise::enterprise::cloud::billings::get_billing_by_org_id(&target_org_id).await {
|
||||
Ok(Some(b)) if matches!(b.provider, MeteringProvider::NoOp) => {}
|
||||
Ok(_) => {
|
||||
return MetaHttpResponse::not_found(
|
||||
"No active external contract found for this organization",
|
||||
);
|
||||
}
|
||||
Err(e) => {
|
||||
return MetaHttpResponse::internal_error(e.to_string());
|
||||
}
|
||||
};
|
||||
|
||||
match o2_enterprise::enterprise::cloud::customer_billings::delete_by_org_id(&target_org_id)
|
||||
.await
|
||||
{
|
||||
Ok(_) => MetaHttpResponse::json(serde_json::json!({"status": "ok"})),
|
||||
Err(e) => {
|
||||
MetaHttpResponse::internal_error(format!("Failed to revoke external contract: {e}"))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// RenameOrganization
|
||||
#[utoipa::path(
|
||||
put,
|
||||
|
||||
@@ -966,6 +966,15 @@ pub fn service_routes() -> Router {
|
||||
"/{org_id}/extend_trial_period",
|
||||
put(organization::org::extend_trial_period),
|
||||
)
|
||||
.route(
|
||||
"/{org_id}/external_contract",
|
||||
post(organization::org::create_external_contract)
|
||||
.put(organization::org::extend_external_contract),
|
||||
)
|
||||
.route(
|
||||
"/{org_id}/external_contract/{target_org_id}",
|
||||
delete(organization::org::revoke_external_contract),
|
||||
)
|
||||
.route(
|
||||
"/{org_id}/aws-marketplace/link-subscription",
|
||||
post(cloud::aws_marketplace::link_subscription),
|
||||
|
||||
Reference in New Issue
Block a user