CloudWatch Logs: Support Log Anomalies query type (#113067)

pkg/tsdb/cloudwatch/cloudwatch.go

@@ -36,7 +36,7 @@ const (
 	headerFromAlert = "FromAlert"
 
 	defaultRegion = "default"
-	logsQueryMode = "Logs"
+	queryModeLogs = "Logs"
 	// QueryTypes
 	annotationQuery = "annotationQuery"
 	logAction       = "logAction"
@@ -45,7 +45,8 @@ const (
 
 type DataQueryJson struct {
 	dataquery.CloudWatchAnnotationQuery
-	Type string `json:"type,omitempty"`
+	Type     string             `json:"type,omitempty"`
+	LogsMode dataquery.LogsMode `json:"logsMode,omitempty"`
 }
 
 type DataSource struct {
@@ -147,12 +148,22 @@ func (ds *DataSource) QueryData(ctx context.Context, req *backend.QueryDataReque
 	if model.QueryMode != "" {
 		queryMode = string(model.QueryMode)
 	}
-	fromPublicDashboard := model.Type == "" && queryMode == logsQueryMode
-	isSyncLogQuery := ((fromAlert || fromExpression) && queryMode == logsQueryMode) || fromPublicDashboard
+
+	fromPublicDashboard := model.Type == ""
+
+	isLogInsightsQuery := queryMode == queryModeLogs && (model.LogsMode == "" || model.LogsMode == dataquery.LogsModeInsights)
+
+	isSyncLogQuery := isLogInsightsQuery && ((fromAlert || fromExpression) || fromPublicDashboard)
+
 	if isSyncLogQuery {
 		return executeSyncLogQuery(ctx, ds, req)
 	}
 
+	isLogsAnomaliesQuery := model.QueryMode == dataquery.CloudWatchQueryModeLogs && model.LogsMode == dataquery.LogsModeAnomalies
+	if isLogsAnomaliesQuery {
+		return executeLogAnomaliesQuery(ctx, ds, req)
+	}
+
 	var result *backend.QueryDataResponse
 	switch model.Type {
 	case annotationQuery:

pkg/tsdb/cloudwatch/kinds/dataquery/types_dataquery_gen.go

@@ -285,6 +285,13 @@ func NewQueryEditorOperatorValueType() *QueryEditorOperatorValueType {
 	return NewStringOrBoolOrInt64OrArrayOfQueryEditorOperatorType()
 }
 
+type LogsMode string
+
+const (
+	LogsModeInsights  LogsMode = "Insights"
+	LogsModeAnomalies LogsMode = "Anomalies"
+)
+
 type LogsQueryLanguage string
 
 const (
@@ -297,7 +304,9 @@ const (
 type CloudWatchLogsQuery struct {
 	// Whether a query is a Metrics, Logs, or Annotations query
 	QueryMode CloudWatchQueryMode `json:"queryMode"`
-	Id        string              `json:"id"`
+	// Whether a query is a Logs Insights or Logs Anomalies query
+	LogsMode *LogsMode `json:"logsMode,omitempty"`
+	Id       string    `json:"id"`
 	// AWS region to query for the logs
 	Region string `json:"region"`
 	// The CloudWatch Logs Insights query to execute
@@ -347,6 +356,40 @@ func NewLogGroup() *LogGroup {
 	return &LogGroup{}
 }
 
+// Shape of a Cloudwatch Logs Anomalies query
+type CloudWatchLogsAnomaliesQuery struct {
+	Id string `json:"id"`
+	// AWS region to query for the logs
+	Region string `json:"region"`
+	// Whether a query is a Metrics, Logs or Annotations query
+	QueryMode *CloudWatchQueryMode `json:"queryMode,omitempty"`
+	// Whether a query is a Logs Insights or Logs Anomalies query
+	LogsMode *LogsMode `json:"logsMode,omitempty"`
+	// Filter to return only anomalies that are 'SUPPRESSED', 'UNSUPPRESSED', or 'ALL' (default)
+	SuppressionState *string `json:"suppressionState,omitempty"`
+	// A unique identifier for the query within the list of targets.
+	// In server side expressions, the refId is used as a variable name to identify results.
+	// By default, the UI will assign A->Z; however setting meaningful names may be useful.
+	RefId string `json:"refId"`
+	// If hide is set to true, Grafana will filter out the response(s) associated with this query before returning it to the panel.
+	Hide *bool `json:"hide,omitempty"`
+	// Specify the query flavor
+	// TODO make this required and give it a default
+	QueryType *string `json:"queryType,omitempty"`
+	// Used to filter only the anomalies found by a certain anomaly detector
+	AnomalyDetectionARN *string `json:"anomalyDetectionARN,omitempty"`
+	// For mixed data sources the selected datasource is on the query level.
+	// For non mixed scenarios this is undefined.
+	// TODO find a better way to do this ^ that's friendly to schema
+	// TODO this shouldn't be unknown but DataSourceRef | null
+	Datasource any `json:"datasource,omitempty"`
+}
+
+// NewCloudWatchLogsAnomaliesQuery creates a new CloudWatchLogsAnomaliesQuery object.
+func NewCloudWatchLogsAnomaliesQuery() *CloudWatchLogsAnomaliesQuery {
+	return &CloudWatchLogsAnomaliesQuery{}
+}
+
 // Shape of a CloudWatch Annotation query
 // TS type is CloudWatchDefaultQuery = Omit<CloudWatchLogsQuery, 'queryMode'> & CloudWatchMetricsQuery, declared in veneer
 // #CloudWatchDefaultQuery: #CloudWatchLogsQuery & #CloudWatchMetricsQuery @cuetsy(kind="type")

pkg/tsdb/cloudwatch/log_anomalies_query.go

@@ -0,0 +1,146 @@
+package cloudwatch
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs"
+
+	cloudwatchLogsTypes "github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs/types"
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+	"github.com/grafana/grafana-plugin-sdk-go/data"
+	"github.com/grafana/grafana/pkg/tsdb/cloudwatch/kinds/dataquery"
+)
+
+var executeLogAnomaliesQuery = func(ctx context.Context, ds *DataSource, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error) {
+	resp := backend.NewQueryDataResponse()
+
+	for _, q := range req.Queries {
+		var anomaliesQuery dataquery.CloudWatchLogsAnomaliesQuery
+		err := json.Unmarshal(q.JSON, &anomaliesQuery)
+		if err != nil {
+			continue
+		}
+
+		region := anomaliesQuery.Region
+		if region == "" || region == defaultRegion {
+			anomaliesQuery.Region = ds.Settings.Region
+		}
+
+		logsClient, err := ds.getCWLogsClient(ctx, region)
+		if err != nil {
+			return nil, err
+		}
+
+		listAnomaliesInput := &cloudwatchlogs.ListAnomaliesInput{}
+		if anomaliesQuery.SuppressionState != nil {
+			listAnomaliesInput.SuppressionState = getSuppressionState(*anomaliesQuery.SuppressionState)
+		}
+
+		if anomaliesQuery.AnomalyDetectionARN == nil || *anomaliesQuery.AnomalyDetectionARN != "" {
+			listAnomaliesInput.AnomalyDetectorArn = anomaliesQuery.AnomalyDetectionARN
+		}
+
+		response, err := logsClient.ListAnomalies(ctx, listAnomaliesInput)
+
+		if err != nil {
+			result := backend.NewQueryDataResponse()
+			result.Responses[q.RefID] = backend.ErrorResponseWithErrorSource(backend.DownstreamError(fmt.Errorf("%v: %w", "failed to call cloudwatch:ListAnomalies", err)))
+			return result, nil
+		}
+
+		dataframe, err := logsAnomaliesResultsToDataframes(response)
+		if err != nil {
+			return nil, err
+		}
+
+		respD := resp.Responses[q.RefID]
+		respD.Frames = data.Frames{dataframe}
+		resp.Responses[q.RefID] = respD
+	}
+
+	return resp, nil
+
+}
+
+func logsAnomaliesResultsToDataframes(response *cloudwatchlogs.ListAnomaliesOutput) (*data.Frame, error) {
+	frame := data.NewFrame("Log anomalies")
+
+	if len(response.Anomalies) == 0 {
+		return frame, nil
+	}
+
+	n := len(response.Anomalies)
+	anomalyArns := make([]string, n)
+	descriptions := make([]string, n)
+	suppressedStatus := make([]bool, n)
+
+	priorities := make([]string, n)
+	patterns := make([]string, n)
+	statuses := make([]string, n)
+	logGroupArnLists := make([]string, n)
+	firstSeens := make([]time.Time, n)
+	lastSeens := make([]time.Time, n)
+	logTrends := make([]*json.RawMessage, n)
+
+	for i, anomaly := range response.Anomalies {
+		anomalyArns[i] = *anomaly.AnomalyDetectorArn
+		descriptions[i] = *anomaly.Description
+		suppressedStatus[i] = *anomaly.Suppressed
+		priorities[i] = *anomaly.Priority
+		if anomaly.PatternString != nil {
+			patterns[i] = *anomaly.PatternString
+		}
+		statuses[i] = string(anomaly.State)
+		logGroupArnLists[i] = strings.Join(anomaly.LogGroupArnList, ",")
+
+		firstSeens[i] = time.UnixMilli(anomaly.FirstSeen)
+
+		lastSeens[i] = time.UnixMilli(anomaly.LastSeen)
+
+		// data.Frame returned from the backend cannot contain fields of type data.Frames
+		// so histogram is kept as json.RawMessageto be built as sparkline table cell on the FE
+		histogramField := anomaly.Histogram
+		histogramJSON, err := json.Marshal(histogramField)
+		if err != nil {
+			logTrends[i] = nil
+		} else {
+			rawMsg := json.RawMessage(histogramJSON)
+			logTrends[i] = &rawMsg
+		}
+	}
+
+	newFields := make([]*data.Field, 0, len(response.Anomalies))
+
+	newFields = append(newFields, data.NewField("state", nil, statuses).SetConfig(&data.FieldConfig{DisplayName: "State"}))
+	newFields = append(newFields, data.NewField("description", nil, descriptions).SetConfig(&data.FieldConfig{DisplayName: "Anomaly"}))
+	newFields = append(newFields, data.NewField("priority", nil, priorities).SetConfig(&data.FieldConfig{DisplayName: "Priority"}))
+	newFields = append(newFields, data.NewField("patternString", nil, patterns).SetConfig(&data.FieldConfig{DisplayName: "Log Pattern"}))
+	// FE expects the field name to be logTrend in order to identify histogram field for sparkline rendering
+	newFields = append(newFields, data.NewField("logTrend", nil, logTrends).SetConfig(&data.FieldConfig{DisplayName: "Log Trend"}))
+	newFields = append(newFields, data.NewField("firstSeen", nil, firstSeens).SetConfig(&data.FieldConfig{DisplayName: "First seen"}))
+	newFields = append(newFields, data.NewField("lastSeen", nil, lastSeens).SetConfig(&data.FieldConfig{DisplayName: "Last seen"}))
+	newFields = append(newFields, data.NewField("suppressed", nil, suppressedStatus).SetConfig(&data.FieldConfig{DisplayName: "Suppressed?"}))
+	newFields = append(newFields, data.NewField("logGroupArnList", nil, logGroupArnLists).SetConfig(&data.FieldConfig{DisplayName: "Log Groups"}))
+	newFields = append(newFields, data.NewField("anomalyArn", nil, anomalyArns).SetConfig(&data.FieldConfig{DisplayName: "Anomaly Arn"}))
+
+	frame.Fields = newFields
+	setPreferredVisType(frame, data.VisTypeTable)
+	return frame, nil
+}
+
+func getSuppressionState(suppressionState string) cloudwatchLogsTypes.SuppressionState {
+	switch suppressionState {
+	case "suppressed":
+		return cloudwatchLogsTypes.SuppressionStateSuppressed
+	case "unsuppressed":
+		return cloudwatchLogsTypes.SuppressionStateUnsuppressed
+	case "all":
+		return ""
+	default:
+		return ""
+	}
+}

pkg/tsdb/cloudwatch/log_anomalies_query_test.go

@@ -0,0 +1,212 @@
+package cloudwatch
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	cloudwatchLogsTypes "github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs/types"
+
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+	"github.com/grafana/grafana/pkg/tsdb/cloudwatch/models"
+	"github.com/stretchr/testify/assert"
+)
+
+func Test_executeLogAnomaliesQuery(t *testing.T) {
+	origNewCWClient := NewCWClient
+	t.Cleanup(func() {
+		NewCWClient = origNewCWClient
+	})
+
+	var cli fakeCWLogsClient
+	NewCWLogsClient = func(aws.Config) models.CWLogsClient {
+		return &cli
+	}
+
+	t.Run("getCWLogsClient is called with correct suppression state", func(t *testing.T) {
+		testcases := []struct {
+			name                    string
+			suppressionStateInQuery string
+			result                  cloudwatchLogsTypes.SuppressionState
+		}{
+			{
+				"suppressed state",
+				"suppressed",
+				cloudwatchLogsTypes.SuppressionStateSuppressed,
+			},
+			{
+				"unsuppressed state",
+				"unsuppressed",
+				cloudwatchLogsTypes.SuppressionStateUnsuppressed,
+			},
+			{
+				"empty state",
+				"",
+				"",
+			},
+			{
+				"all state",
+				"all",
+				"",
+			},
+		}
+
+		for _, tc := range testcases {
+			t.Run(tc.name, func(t *testing.T) {
+				cli = fakeCWLogsClient{anomalies: []cloudwatchLogsTypes.Anomaly{}}
+				ds := newTestDatasource()
+
+				_, err := ds.QueryData(context.Background(), &backend.QueryDataRequest{
+					Headers:       map[string]string{headerFromAlert: ""},
+					PluginContext: backend.PluginContext{DataSourceInstanceSettings: &backend.DataSourceInstanceSettings{}},
+					Queries: []backend.DataQuery{
+						{
+							TimeRange: backend.TimeRange{From: time.Unix(0, 0), To: time.Unix(1, 0)},
+							JSON: json.RawMessage(`{
+						"queryMode":    "Logs",
+						"logsMode": "Anomalies",
+						"suppressionState": "` + tc.suppressionStateInQuery + `",
+						"region": "us-east-1"
+						}`),
+						},
+					},
+				})
+				assert.NoError(t, err)
+				assert.Equal(t, tc.result, cli.calls.listAnomalies[0].SuppressionState)
+			})
+		}
+	})
+}
+
+func Test_executeLogAnomaliesQuery_returns_data_frames(t *testing.T) {
+	origNewCWClient := NewCWClient
+	t.Cleanup(func() {
+		NewCWClient = origNewCWClient
+	})
+
+	var cli fakeCWLogsClient
+	NewCWLogsClient = func(aws.Config) models.CWLogsClient {
+		return &cli
+	}
+	t.Run("returns log anomalies data frames", func(t *testing.T) {
+		cli = fakeCWLogsClient{anomalies: []cloudwatchLogsTypes.Anomaly{
+			{
+				AnomalyId:          aws.String("anomaly-1"),
+				AnomalyDetectorArn: aws.String("arn:aws:logs:us-east-1:123456789012:anomaly-detector:anomaly-detector-1"),
+				FirstSeen:          1622505600000, // June 1, 2021 00:00:00 GMT
+				LastSeen:           1622592000000, // June 2, 2021 00:00:00 GMT
+				LogGroupArnList:    []string{"arn:aws:logs:us-east-1:1234567:log-group-1:id-1", "arn:aws:logs:us-east-1:1234567:log-group-2:id-2"},
+				Description:        aws.String("Description 1"),
+				State:              cloudwatchLogsTypes.StateActive,
+				Priority:           aws.String("high"),
+				PatternString:      aws.String(`{"ClusterName":"PetSite","Namespace":"default","Service":"service-petsite",,"instance":"instance"-5:Token-6,"job":"kubernetes-service-endpoints","pod_name":"pod_name"-9,"prom_metric_type":"counter"}`),
+				Suppressed:         aws.Bool(false),
+				Histogram: map[string]int64{
+					"1622505600000": 5,
+					"1622519200000": 10,
+					"1622532800000": 7,
+				},
+			},
+			{
+				AnomalyId:          aws.String("anomaly-2"),
+				AnomalyDetectorArn: aws.String("arn:aws:logs:us-east-1:123456789012:anomaly-detector:anomaly-detector-2"),
+				FirstSeen:          1622592000000, // June 2, 2021 00:00:00 GMT
+				LastSeen:           1622678400000, // June 3, 2021 00:00:00 GMT
+				LogGroupArnList:    []string{"arn:aws:logs:us-east-1:1234567:log-group-1:id-3", "arn:aws:logs:us-east-1:1234567:log-group-2:id-4"},
+				Description:        aws.String("Description 2"),
+				State:              cloudwatchLogsTypes.StateSuppressed,
+				Priority:           aws.String("low"),
+				PatternString:      aws.String(`{"ClusterName":"PetSite","Namespace":"default","Service":"service-petsite","dotnet_collection_count_total":"dotnet_collection_count_total"-3}`),
+				Suppressed:         aws.Bool(true),
+				Histogram: map[string]int64{
+					"1622592000000": 3,
+				},
+			},
+		}}
+
+		ds := newTestDatasource()
+
+		resp, err := ds.QueryData(context.Background(), &backend.QueryDataRequest{
+			Headers:       map[string]string{headerFromAlert: ""},
+			PluginContext: backend.PluginContext{DataSourceInstanceSettings: &backend.DataSourceInstanceSettings{}},
+			Queries: []backend.DataQuery{
+				{
+					TimeRange: backend.TimeRange{From: time.Unix(0, 0), To: time.Unix(1, 0)},
+					JSON: json.RawMessage(`{
+							"queryMode":    "Logs",
+							"logsMode": "Anomalies",
+							"suppressionState": "all",
+							"region": "us-east-1"
+							}`),
+				},
+			},
+		})
+
+		assert.NoError(t, err)
+		assert.Len(t, resp.Responses, 1)
+		for _, r := range resp.Responses {
+			assert.Len(t, r.Frames, 1)
+			frame := r.Frames[0]
+			assert.Equal(t, "Log anomalies", frame.Name)
+			assert.Len(t, frame.Fields, 10)
+
+			stateField := frame.Fields[0]
+			assert.Equal(t, "state", stateField.Name)
+			assert.Equal(t, "Active", stateField.At(0))
+			assert.Equal(t, "Suppressed", stateField.At(1))
+
+			descriptionField := frame.Fields[1]
+			assert.Equal(t, "description", descriptionField.Name)
+			assert.Equal(t, "Description 1", descriptionField.At(0))
+			assert.Equal(t, "Description 2", descriptionField.At(1))
+
+			priorityField := frame.Fields[2]
+			assert.Equal(t, "priority", priorityField.Name)
+			assert.Equal(t, "high", priorityField.At(0))
+			assert.Equal(t, "low", priorityField.At(1))
+
+			patternStringField := frame.Fields[3]
+			assert.Equal(t, "patternString", patternStringField.Name)
+			assert.Equal(t, `{"ClusterName":"PetSite","Namespace":"default","Service":"service-petsite",,"instance":"instance"-5:Token-6,"job":"kubernetes-service-endpoints","pod_name":"pod_name"-9,"prom_metric_type":"counter"}`, patternStringField.At(0))
+			assert.Equal(t, `{"ClusterName":"PetSite","Namespace":"default","Service":"service-petsite","dotnet_collection_count_total":"dotnet_collection_count_total"-3}`, patternStringField.At(1))
+
+			histogramField := frame.Fields[4]
+			assert.Equal(t, "logTrend", histogramField.Name)
+
+			histogram0 := histogramField.At(0).(*json.RawMessage)
+			var histData0 map[string]int64
+			err = json.Unmarshal(*histogram0, &histData0)
+			assert.NoError(t, err)
+			assert.Equal(t, int64(5), histData0["1622505600000"])
+			assert.Equal(t, int64(10), histData0["1622519200000"])
+			assert.Equal(t, int64(7), histData0["1622532800000"])
+
+			firstSeenField := frame.Fields[5]
+			assert.Equal(t, "firstSeen", firstSeenField.Name)
+			assert.Equal(t, time.Unix(1622505600, 0), firstSeenField.At(0))
+			assert.Equal(t, time.Unix(1622592000, 0), firstSeenField.At(1))
+
+			lastSeenField := frame.Fields[6]
+			assert.Equal(t, "lastSeen", lastSeenField.Name)
+			assert.Equal(t, time.Unix(1622592000, 0), lastSeenField.At(0))
+			assert.Equal(t, time.Unix(1622678400, 0), lastSeenField.At(1))
+
+			suppressedField := frame.Fields[7]
+			assert.Equal(t, "suppressed", suppressedField.Name)
+			assert.Equal(t, false, suppressedField.At(0))
+			assert.Equal(t, true, suppressedField.At(1))
+
+			logGroupArnListField := frame.Fields[8]
+			assert.Equal(t, "logGroupArnList", logGroupArnListField.Name)
+			assert.Equal(t, "arn:aws:logs:us-east-1:1234567:log-group-1:id-1,arn:aws:logs:us-east-1:1234567:log-group-2:id-2", logGroupArnListField.At(0))
+			assert.Equal(t, "arn:aws:logs:us-east-1:1234567:log-group-1:id-3,arn:aws:logs:us-east-1:1234567:log-group-2:id-4", logGroupArnListField.At(1))
+
+			anomalyDetectorArnField := frame.Fields[9]
+			assert.Equal(t, "anomalyArn", anomalyDetectorArnField.Name)
+			assert.Equal(t, "arn:aws:logs:us-east-1:123456789012:anomaly-detector:anomaly-detector-1", anomalyDetectorArnField.At(0))
+			assert.Equal(t, "arn:aws:logs:us-east-1:123456789012:anomaly-detector:anomaly-detector-2", anomalyDetectorArnField.At(1))
+		}
+	})
+}

pkg/tsdb/cloudwatch/log_sync_query_test.go

@@ -137,7 +137,7 @@ func Test_executeSyncLogQuery(t *testing.T) {
 		executeSyncLogQuery = origExecuteSyncLogQuery
 	})
 
-	t.Run("when query mode is 'Logs' and does not include type or subtype", func(t *testing.T) {
+	t.Run("when query mode is 'Logs Insights' and does not include type or subtype", func(t *testing.T) {
 		origExecuteSyncLogQuery := executeSyncLogQuery
 		syncCalled := false
 		executeSyncLogQuery = func(ctx context.Context, e *DataSource, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error) {

pkg/tsdb/cloudwatch/mocks/logs.go

@@ -66,3 +66,9 @@ func (m *MockLogEvents) GetLogEvents(ctx context.Context, input *cloudwatchlogs.
 
 	return args.Get(0).(*cloudwatchlogs.GetLogEventsOutput), args.Error(1)
 }
+
+func (m *MockLogEvents) ListAnomalies(ctx context.Context, input *cloudwatchlogs.ListAnomaliesInput, optFns ...func(*cloudwatchlogs.Options)) (*cloudwatchlogs.ListAnomaliesOutput, error) {
+	args := m.Called(ctx, input, optFns)
+
+	return args.Get(0).(*cloudwatchlogs.ListAnomaliesOutput), args.Error(1)
+}

pkg/tsdb/cloudwatch/models/api.go

@@ -76,6 +76,7 @@ type CWLogsClient interface {
 
 	cloudwatchlogs.GetLogEventsAPIClient
 	cloudwatchlogs.DescribeLogGroupsAPIClient
+	cloudwatchlogs.ListAnomaliesAPIClient
 }
 
 type CWClient interface {

pkg/tsdb/cloudwatch/test_utils.go

@@ -29,12 +29,15 @@ type fakeCWLogsClient struct {
 	queryResults   cloudwatchlogs.GetQueryResultsOutput
 
 	logGroupsIndex int
+
+	anomalies []cloudwatchlogstypes.Anomaly
 }
 
 type logsQueryCalls struct {
 	startQuery        []*cloudwatchlogs.StartQueryInput
 	getEvents         []*cloudwatchlogs.GetLogEventsInput
 	describeLogGroups []*cloudwatchlogs.DescribeLogGroupsInput
+	listAnomalies     []*cloudwatchlogs.ListAnomaliesInput
 }
 
 func (m *fakeCWLogsClient) GetQueryResults(_ context.Context, _ *cloudwatchlogs.GetQueryResultsInput, _ ...func(*cloudwatchlogs.Options)) (*cloudwatchlogs.GetQueryResultsOutput, error) {
@@ -55,6 +58,14 @@ func (m *fakeCWLogsClient) StopQuery(_ context.Context, _ *cloudwatchlogs.StopQu
 	}, nil
 }
 
+func (m *fakeCWLogsClient) ListAnomalies(_ context.Context, input *cloudwatchlogs.ListAnomaliesInput, _ ...func(*cloudwatchlogs.Options)) (*cloudwatchlogs.ListAnomaliesOutput, error) {
+	m.calls.listAnomalies = append(m.calls.listAnomalies, input)
+
+	return &cloudwatchlogs.ListAnomaliesOutput{
+		Anomalies: m.anomalies,
+	}, nil
+}
+
 type mockLogsSyncClient struct {
 	mock.Mock
 }
@@ -80,6 +91,11 @@ func (m *mockLogsSyncClient) StartQuery(ctx context.Context, input *cloudwatchlo
 	return args.Get(0).(*cloudwatchlogs.StartQueryOutput), args.Error(1)
 }
 
+func (m *mockLogsSyncClient) ListAnomalies(ctx context.Context, input *cloudwatchlogs.ListAnomaliesInput, optFns ...func(*cloudwatchlogs.Options)) (*cloudwatchlogs.ListAnomaliesOutput, error) {
+	args := m.Called(ctx, input, optFns)
+	return args.Get(0).(*cloudwatchlogs.ListAnomaliesOutput), args.Error(1)
+}
+
 func (m *fakeCWLogsClient) DescribeLogGroups(_ context.Context, input *cloudwatchlogs.DescribeLogGroupsInput, _ ...func(*cloudwatchlogs.Options)) (*cloudwatchlogs.DescribeLogGroupsOutput, error) {
 	m.calls.describeLogGroups = append(m.calls.describeLogGroups, input)
 	output := &m.logGroups[m.logGroupsIndex]