test: add home and dataset metadata api

tests/integration/dataset_metadata_test.go

@@ -0,0 +1,92 @@
+// Copyright (c) 2023 Cloudnatively Services Pvt Ltd
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+package main
+
+import (
+	"bytes"
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+type prismDatasetMetadata struct {
+	Tags []string `json:"tags"`
+}
+
+type prismTaggedDataset struct {
+	Name      string `json:"name"`
+	LogSource string `json:"logSource"`
+}
+
+func putPrismDatasetTags(t *testing.T, dataset string, tags []string) prismDatasetMetadata {
+	t.Helper()
+	body, err := json.Marshal(map[string][]string{"tags": tags})
+	require.NoError(t, err)
+	req, err := NewGlob.QueryClient.NewRequestAtPath("PUT", "api/prism/v1/datasets/"+dataset, bytes.NewReader(body))
+	require.NoError(t, err)
+	response, err := NewGlob.QueryClient.Do(req)
+	require.NoError(t, err)
+	defer response.Body.Close()
+	responseBody := readAsString(response.Body)
+	require.Equalf(t, 200, response.StatusCode, "PUT dataset metadata returned %s: %s", response.Status, responseBody)
+	var metadata prismDatasetMetadata
+	require.NoError(t, json.Unmarshal([]byte(responseBody), &metadata))
+	return metadata
+}
+
+func listPrismDatasetsByTag(t *testing.T, tag string) []prismTaggedDataset {
+	t.Helper()
+	req, err := NewGlob.QueryClient.NewRequestAtPath("GET", "api/prism/v1/datasets/tags/"+tag, nil)
+	require.NoError(t, err)
+	response, err := NewGlob.QueryClient.Do(req)
+	require.NoError(t, err)
+	defer response.Body.Close()
+	require.Equalf(t, 200, response.StatusCode, "GET datasets by tag returned %s", response.Status)
+	datasets, err := readJsonBody[[]prismTaggedDataset](response.Body)
+	require.NoError(t, err)
+	return datasets
+}
+
+func TestEnterprisePrismDatasetTags(t *testing.T) {
+	if NewGlob.Edition != "enterprise" {
+		t.Skip("Prism dataset metadata endpoints are only available in Enterprise")
+	}
+	t.Parallel()
+
+	const tag = "agent-observability"
+	datasetA := NewGlob.Stream + "agentmetadataa"
+	datasetB := NewGlob.Stream + "agentmetadatab"
+	traceHeaders := map[string]string{
+		"X-P-Telemetry-Type": "traces",
+		"X-P-Log-Source":     "otel-traces",
+	}
+	CreateStreamWithHeader(t, NewGlob.QueryClient, datasetA, traceHeaders)
+	t.Cleanup(func() { DeleteStream(t, NewGlob.PBClient, datasetA) })
+	CreateStreamWithHeader(t, NewGlob.QueryClient, datasetB, traceHeaders)
+	t.Cleanup(func() { DeleteStream(t, NewGlob.PBClient, datasetB) })
+
+	for _, dataset := range []string{datasetA, datasetB} {
+		metadata := putPrismDatasetTags(t, dataset, []string{tag})
+		require.Equal(t, []string{tag}, metadata.Tags)
+	}
+
+	tagged := listPrismDatasetsByTag(t, tag)
+	require.Contains(t, tagged, prismTaggedDataset{Name: datasetA, LogSource: "otel-traces"})
+	require.Contains(t, tagged, prismTaggedDataset{Name: datasetB, LogSource: "otel-traces"})
+
+	metadata := putPrismDatasetTags(t, datasetA, []string{})
+	require.NotNil(t, metadata.Tags)
+	require.Empty(t, metadata.Tags)
+
+	tagged = listPrismDatasetsByTag(t, tag)
+	for _, dataset := range tagged {
+		require.NotEqual(t, datasetA, dataset.Name)
+	}
+	require.Contains(t, tagged, prismTaggedDataset{Name: datasetB, LogSource: "otel-traces"})
+}

tests/integration/query_test.go

@@ -30,6 +30,39 @@ type queryWithFieldsResponse struct {
 	Records []json.RawMessage `json:"records"`
 }
 
+type prismHomeResponse struct {
+	Datasets  []prismHomeDataset `json:"datasets"`
+	Checklist prismHomeChecklist `json:"checklist"`
+	LLMKeys   []json.RawMessage   `json:"llmKeys"`
+}
+
+type prismHomeDataset struct {
+	Title         string   `json:"title"`
+	DatasetType   string   `json:"datasetType"`
+	TimePartition *string  `json:"timePartition"`
+	DatasetFormat string   `json:"datasetFormat"`
+	Ingestion     bool     `json:"ingestion"`
+	Tags          []string `json:"tags"`
+	Labels        []string `json:"labels"`
+}
+
+type prismHomeChecklist struct {
+	DataIngested    *bool `json:"dataIngested"`
+	KeystoneCreated *bool `json:"keystoneCreated"`
+	AlertCreated    *bool `json:"alertCreated"`
+	UserAdded       *bool `json:"userAdded"`
+}
+
+type prismHomeSearchResponse struct {
+	Resources []prismHomeSearchResource `json:"resources"`
+}
+
+type prismHomeSearchResource struct {
+	ID           string `json:"id"`
+	Name         string `json:"name"`
+	ResourceType string `json:"resource_type"`
+}
+
 type countsResponse struct {
 	Fields  []string       `json:"fields"`
 	Records []countsRecord `json:"records"`
@@ -83,6 +116,62 @@ func TestSmokeIngestAndQuery(t *testing.T) {
 	// the same batch window.
 	time.Sleep(120 * time.Second)
 
+	t.Run("PrismHome", func(t *testing.T) {
+		req, err := NewGlob.QueryClient.NewRequestAtPath("GET", "api/prism/v1/home", nil)
+		require.NoError(t, err)
+		response, err := NewGlob.QueryClient.Do(req)
+		require.NoError(t, err)
+		defer response.Body.Close()
+		require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s", response.Status)
+
+		home, err := readJsonBody[prismHomeResponse](response.Body)
+		require.NoError(t, err)
+		require.NotNil(t, home.Checklist.DataIngested)
+		require.NotNil(t, home.Checklist.KeystoneCreated)
+		require.NotNil(t, home.Checklist.AlertCreated)
+		require.NotNil(t, home.Checklist.UserAdded)
+		require.True(t, *home.Checklist.DataIngested)
+		if NewGlob.Edition == "enterprise" {
+			require.NotNil(t, home.LLMKeys)
+		}
+
+		var dataset *prismHomeDataset
+		for i := range home.Datasets {
+			if home.Datasets[i].Title == stream1 {
+				dataset = &home.Datasets[i]
+				break
+			}
+		}
+		require.NotNil(t, dataset, "home response did not include stream %s", stream1)
+		require.Equal(t, "logs", dataset.DatasetType)
+		require.Nil(t, dataset.TimePartition)
+		require.Equal(t, "json", dataset.DatasetFormat)
+		require.True(t, dataset.Ingestion)
+		require.Empty(t, dataset.Tags)
+		require.Empty(t, dataset.Labels)
+	})
+
+	t.Run("PrismHomeSearch", func(t *testing.T) {
+		req, err := NewGlob.QueryClient.NewRequestAtPath("GET", "api/prism/v1/home/search", nil)
+		require.NoError(t, err)
+		params := req.URL.Query()
+		params.Set("key", stream1)
+		req.URL.RawQuery = params.Encode()
+
+		response, err := NewGlob.QueryClient.Do(req)
+		require.NoError(t, err)
+		defer response.Body.Close()
+		require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s", response.Status)
+
+		search, err := readJsonBody[prismHomeSearchResponse](response.Body)
+		require.NoError(t, err)
+		require.Contains(t, search.Resources, prismHomeSearchResource{
+			ID:           stream1,
+			Name:         stream1,
+			ResourceType: "DataSet",
+		})
+	})
+
 	t.Run("IngestEventsToStream", func(t *testing.T) {
 		t.Parallel()
 		QueryLogStreamCount(t, NewGlob.PBClient, stream1, 50)