influxdb/router/tests/grpc.rs

use std::time::Duration;

use assert_matches::assert_matches;
use generated_types::influxdata::iox::namespace::v1::{
    namespace_service_server::NamespaceService, *,
};
use hyper::StatusCode;
use iox_time::{SystemProvider, TimeProvider};
use router::{
    dml_handlers::{DmlError, RetentionError},
    namespace_resolver::{self, NamespaceCreationError},
    server::http::Error,
};
use tonic::{Code, Request};

use crate::common::TestContext;

pub mod common;

/// Ensure invoking the gRPC NamespaceService to create a namespace populates
/// the catalog.
#[tokio::test]
async fn test_namespace_create() {
    // Initialise a TestContext requiring explicit namespace creation.
    let ctx = TestContext::new(false, None).await;

    // Try writing to the non-existant namespace, which should return an error.
    let now = SystemProvider::default()
        .now()
        .timestamp_nanos()
        .to_string();
    let lp = "platanos,tag1=A,tag2=B val=42i ".to_string() + &now;

    let response = ctx
        .write_lp("bananas", "test", &lp)
        .await
        .expect_err("write failed");

    assert_matches!(
        response,
        Error::NamespaceResolver(namespace_resolver::Error::Create(
            NamespaceCreationError::Reject(_)
        ))
    );
    assert_eq!(
        response.to_string(),
        "rejecting write due to non-existing namespace: bananas_test"
    );

    // The failed write MUST NOT populate the catalog.
    {
        let current = ctx
            .catalog()
            .repositories()
            .await
            .namespaces()
            .list()
            .await
            .expect("failed to query for existing namespaces");
        assert!(current.is_empty());
    }

    // The RPC endpoint must know nothing about the namespace either.
    {
        let current = ctx
            .grpc_delegate()
            .namespace_service()
            .get_namespaces(Request::new(Default::default()))
            .await
            .expect("must return namespaces")
            .into_inner();
        assert!(current.namespaces.is_empty());
    }

    const RETENTION: i64 = Duration::from_secs(42 * 60 * 60).as_nanos() as _;

    // Explicitly create the namespace.
    let req = CreateNamespaceRequest {
        name: "bananas_test".to_string(),
        retention_period_ns: Some(RETENTION),
    };
    let got = ctx
        .grpc_delegate()
        .namespace_service()
        .create_namespace(Request::new(req))
        .await
        .expect("failed to create namespace")
        .into_inner()
        .namespace
        .expect("no namespace in response");

    assert_eq!(got.name, "bananas_test");
    assert_eq!(got.id, 1);
    assert_eq!(got.retention_period_ns, Some(RETENTION));

    // The list namespace RPC should show the new namespace
    {
        let list = ctx
            .grpc_delegate()
            .namespace_service()
            .get_namespaces(Request::new(Default::default()))
            .await
            .expect("must return namespaces")
            .into_inner();
        assert_matches!(list.namespaces.as_slice(), [ns] => {
            assert_eq!(*ns, got);
        });
    }

    // The catalog should contain the namespace.
    {
        let db_list = ctx
            .catalog()
            .repositories()
            .await
            .namespaces()
            .list()
            .await
            .expect("query failure");
        assert_matches!(db_list.as_slice(), [ns] => {
            assert_eq!(ns.id.get(), got.id);
            assert_eq!(ns.name, got.name);
            assert_eq!(ns.retention_period_ns, got.retention_period_ns);
        });
    }

    // And writing should succeed
    let response = ctx
        .write_lp("bananas", "test", lp)
        .await
        .expect("write failed");
    assert_eq!(response.status(), StatusCode::NO_CONTENT);
}

/// Ensure creating a namespace with a retention period of 0 maps to "infinite"
/// and not "none".
#[tokio::test]
async fn test_create_namespace_0_retention_period() {
    // Initialise a TestContext requiring explicit namespace creation.
    let ctx = TestContext::new(false, None).await;

    // Explicitly create the namespace.
    let req = CreateNamespaceRequest {
        name: "bananas_test".to_string(),
        retention_period_ns: Some(0), // A zero!
    };
    let got = ctx
        .grpc_delegate()
        .namespace_service()
        .create_namespace(Request::new(req))
        .await
        .expect("failed to create namespace")
        .into_inner()
        .namespace
        .expect("no namespace in response");

    assert_eq!(got.name, "bananas_test");
    assert_eq!(got.id, 1);
    assert_eq!(got.retention_period_ns, None);

    // The list namespace RPC should show the new namespace
    {
        let list = ctx
            .grpc_delegate()
            .namespace_service()
            .get_namespaces(Request::new(Default::default()))
            .await
            .expect("must return namespaces")
            .into_inner();
        assert_matches!(list.namespaces.as_slice(), [ns] => {
            assert_eq!(*ns, got);
        });
    }

    // The catalog should contain the namespace.
    {
        let db_list = ctx
            .catalog()
            .repositories()
            .await
            .namespaces()
            .list()
            .await
            .expect("query failure");
        assert_matches!(db_list.as_slice(), [ns] => {
            assert_eq!(ns.id.get(), got.id);
            assert_eq!(ns.name, got.name);
            assert_eq!(ns.retention_period_ns, got.retention_period_ns);
        });
    }

    // And writing should succeed
    let response = ctx
        .write_lp("bananas", "test", "platanos,tag1=A,tag2=B val=42i 42424242")
        .await
        .expect("write failed");
    assert_eq!(response.status(), StatusCode::NO_CONTENT);
}

/// Ensure creating a namespace with a negative retention period is rejected.
#[tokio::test]
async fn test_create_namespace_negative_retention_period() {
    // Initialise a TestContext requiring explicit namespace creation.
    let ctx = TestContext::new(false, None).await;

    // Explicitly create the namespace.
    let req = CreateNamespaceRequest {
        name: "bananas_test".to_string(),
        retention_period_ns: Some(-42),
    };
    let err = ctx
        .grpc_delegate()
        .namespace_service()
        .create_namespace(Request::new(req))
        .await
        .expect_err("negative retention period should fail to create namespace");

    assert_eq!(err.code(), Code::InvalidArgument);
    assert_eq!(err.message(), "invalid negative retention period");

    // The list namespace RPC should not show a new namespace
    {
        let list = ctx
            .grpc_delegate()
            .namespace_service()
            .get_namespaces(Request::new(Default::default()))
            .await
            .expect("must return namespaces")
            .into_inner();
        assert!(list.namespaces.is_empty());
    }

    // The catalog should not contain the namespace.
    {
        let db_list = ctx
            .catalog()
            .repositories()
            .await
            .namespaces()
            .list()
            .await
            .expect("query failure");
        assert!(db_list.is_empty());
    }

    // And writing should not succeed
    let response = ctx
        .write_lp("bananas", "test", "platanos,tag1=A,tag2=B val=42i 42424242")
        .await
        .expect_err("write should fail");
    assert_matches!(
        response,
        Error::NamespaceResolver(namespace_resolver::Error::Create(
            NamespaceCreationError::Reject(_)
        ))
    );
}

/// Ensure updating a namespace with a retention period of 0 maps to "infinite"
/// and not "none".
#[tokio::test]
async fn test_update_namespace_0_retention_period() {
    // Initialise a TestContext requiring explicit namespace creation.
    let ctx = TestContext::new(false, None).await;

    // Explicitly create the namespace.
    let create = ctx
        .grpc_delegate()
        .namespace_service()
        .create_namespace(Request::new(CreateNamespaceRequest {
            name: "bananas_test".to_string(),
            retention_period_ns: Some(42),
        }))
        .await
        .expect("failed to create namespace")
        .into_inner()
        .namespace
        .expect("no namespace in response");

    // And writing in the past should fail
    ctx.write_lp("bananas", "test", "platanos,tag1=A,tag2=B val=42i 42424242")
        .await
        .expect_err("write outside retention period should fail");

    let got = ctx
        .grpc_delegate()
        .namespace_service()
        .update_namespace_retention(Request::new(UpdateNamespaceRetentionRequest {
            name: "bananas_test".to_string(),
            retention_period_ns: Some(0), // A zero!
        }))
        .await
        .expect("failed to create namespace")
        .into_inner()
        .namespace
        .expect("no namespace in response");

    assert_eq!(got.name, create.name);
    assert_eq!(got.id, create.id);
    assert_eq!(create.retention_period_ns, Some(42));
    assert_eq!(got.retention_period_ns, None);

    // The list namespace RPC should show the updated namespace
    {
        let list = ctx
            .grpc_delegate()
            .namespace_service()
            .get_namespaces(Request::new(Default::default()))
            .await
            .expect("must return namespaces")
            .into_inner();
        assert_matches!(list.namespaces.as_slice(), [ns] => {
            assert_eq!(*ns, got);
        });
    }

    // The catalog should contain the namespace.
    {
        let db_list = ctx
            .catalog()
            .repositories()
            .await
            .namespaces()
            .list()
            .await
            .expect("query failure");
        assert_matches!(db_list.as_slice(), [ns] => {
            assert_eq!(ns.id.get(), got.id);
            assert_eq!(ns.name, got.name);
            assert_eq!(ns.retention_period_ns, got.retention_period_ns);
        });
    }

    // The cached entry is not affected, and writes continue to be validated
    // against the old value.
    //
    // https://github.com/influxdata/influxdb_iox/issues/6175

    let err = ctx
        .write_lp("bananas", "test", "platanos,tag1=A,tag2=B val=42i 42424242")
        .await
        .expect_err("cached entry rejects write");

    assert_matches!(
        err,
        router::server::http::Error::DmlHandler(DmlError::Retention(
            RetentionError::OutsideRetention(name)
        )) => {
            assert_eq!(name, "platanos");
        }
    );
}

/// Ensure updating a namespace with a negative retention period fails.
#[tokio::test]
async fn test_update_namespace_negative_retention_period() {
    // Initialise a TestContext requiring explicit namespace creation.
    let ctx = TestContext::new(false, None).await;

    // Explicitly create the namespace.
    let create = ctx
        .grpc_delegate()
        .namespace_service()
        .create_namespace(Request::new(CreateNamespaceRequest {
            name: "bananas_test".to_string(),
            retention_period_ns: Some(42),
        }))
        .await
        .expect("failed to create namespace")
        .into_inner()
        .namespace
        .expect("no namespace in response");

    let err = ctx
        .grpc_delegate()
        .namespace_service()
        .update_namespace_retention(Request::new(UpdateNamespaceRetentionRequest {
            name: "bananas_test".to_string(),
            retention_period_ns: Some(-42),
        }))
        .await
        .expect_err("negative retention period should fail to create namespace");

    assert_eq!(err.code(), Code::InvalidArgument);
    assert_eq!(err.message(), "invalid negative retention period");

    // The list namespace RPC should show the original namespace
    {
        let list = ctx
            .grpc_delegate()
            .namespace_service()
            .get_namespaces(Request::new(Default::default()))
            .await
            .expect("must return namespaces")
            .into_inner();
        assert_matches!(list.namespaces.as_slice(), [ns] => {
            assert_eq!(*ns, create);
        });
    }

    // The catalog should contain the original namespace.
    {
        let db_list = ctx
            .catalog()
            .repositories()
            .await
            .namespaces()
            .list()
            .await
            .expect("query failure");
        assert_matches!(db_list.as_slice(), [ns] => {
            assert_eq!(ns.id.get(), create.id);
            assert_eq!(ns.name, create.name);
            assert_eq!(ns.retention_period_ns, create.retention_period_ns);
        });
    }
}