Added support for filtering reporting resources for logs

preet-dev · preet-dev · commit 680efacfe141 · 2023-05-20T17:06:29.000+05:30
diff --git a/example-definitions/json/log-definition.json b/example-definitions/json/log-definition.json
@@ -45,6 +45,15 @@
         "event.reason": "roundRobin([\"Pulling\", \"Pulled\", \"Created\", \"Pulling\", \"Backoff\"])",
         "event.message": "roundRobin([\"Pulling image k8s.gcr.io/echoserver:1.8\", \"Image downloaded\", \"Created container\", \"Pulling image cjknsjc/ccsdc:fff\", \"Error: ImagePullBackoff\"])"
       }
+    },
+    {
+      "severityOrderFunction": "severityDistributionCount([\"ERROR\", \"WARN\", \"DEBUG\"], [1, 1, 4])",
+      "payloadFrequencySeconds": 20,
+      "payloadCount": 2,
+      "copyCount": 200,
+      "filteredReportingResources": {
+        "node": ["k8s.cluster.name=cluster-skyrim"]
+      }
     }
   ]
 }
diff --git a/example-definitions/qa/log-definition.yaml b/example-definitions/qa/log-definition.yaml
@@ -47,3 +47,9 @@ logs:
       event.domain: 'roundRobin(["k8s"])'
       event.reason: 'roundRobin(["Pulling", "Pulled", "Created", "Pulling", "Backoff"])'
       event.message: 'roundRobin(["Pulling image k8s.gcr.io/echoserver:1.8", "Image downloaded", "Created container", "Pulling image cjknsjc/ccsdc:fff", "Error: ImagePullBackoff"])'
+  - severityOrderFunction: 'severityDistributionCount(["ERROR", "WARN", "DEBUG"], [1, 1, 4])'
+    payloadFrequencySeconds: 20
+    payloadCount: 2
+    copyCount: 200
+    filteredReportingResources:
+      node: ["k8s.cluster.name=cluster-skyrim"]
diff --git a/src/main/java/io/opentelemetry/contrib/generator/telemetry/logs/LogGeneratorThread.java b/src/main/java/io/opentelemetry/contrib/generator/telemetry/logs/LogGeneratorThread.java
@@ -34,6 +34,7 @@
 import jakarta.el.ELProcessor;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections4.MapUtils;
 
 import static io.opentelemetry.contrib.generator.telemetry.misc.GeneratorUtils.*;
 
@@ -72,10 +73,21 @@ public void run() {
             List<ResourceLogs> resourceLogsList = new ArrayList<>();
             ResourceLogs resourceLog;
             LogRecord.Builder partialLogRecord = getLog(logDefinition);
-            for (Map.Entry<String, Integer> reportingResource : logDefinition.getReportingResourcesCounts().entrySet()) {
-                List<Resource> postToResources = getResourceSubsetByPostCount(reportingResource.getKey(), reportingResource.getValue());
-                log.debug(requestID + ": Preparing " + postToResources.size() + " resource logs packets for " + reportingResource);
-                for (Resource eachResource: postToResources) {
+            Map<String, List<Resource>> reportingResourcesByType = new HashMap<>();
+            for (Map.Entry<String, Integer> resourceTypeWithCount:
+                    MapUtils.emptyIfNull(logDefinition.getReportingResourcesCounts()).entrySet()) {
+                reportingResourcesByType.put(resourceTypeWithCount.getKey(),
+                        getResourceSubsetByPostCount(resourceTypeWithCount.getKey(), resourceTypeWithCount.getValue()));
+            }
+            for (Map.Entry<String, Map<String, String>> resourceTypeWithFilter:
+                    MapUtils.emptyIfNull(logDefinition.getParsedFilteredReportingResources()).entrySet()) {
+                reportingResourcesByType.put(resourceTypeWithFilter.getKey(),
+                        getFilteredResources(resourceTypeWithFilter.getKey(), resourceTypeWithFilter.getValue()));
+            }
+            for (Map.Entry<String, List<Resource>> reportingResourceByType : reportingResourcesByType.entrySet()) {
+                log.debug(requestID + ": Preparing " + reportingResourceByType.getValue().size() +
+                        " resource logs packets for " + reportingResourceByType.getKey());
+                for (Resource eachResource: reportingResourceByType.getValue()) {
                     LogRecord logRecord = partialLogRecord.clone().addAllAttributes(getResourceAttributes(logDefinition
                             .getCopyResourceAttributes(), eachResource)).build();
                     List<LogRecord> otelLogs = Collections.nCopies(logDefinition.getCopyCount(), logRecord);
@@ -91,13 +103,15 @@ public void run() {
                             .build();
                     resourceLogsList.add(resourceLog);
                 }
-                log.info(requestID + ": Sending payload for: " + reportingResource);
+                log.info(requestID + ": Sending payload for: " + reportingResourceByType.getKey());
                 ExportLogsServiceRequest resourceLogs = ExportLogsServiceRequest.newBuilder().addAllResourceLogs(resourceLogsList).build();
                 boolean responseStatus = payloadHandler.postPayload(resourceLogs);
                 if (logGeneratorState.getTransportStorage() != null) {
-                    logGeneratorState.getTransportStorage().store(logDefinition.getId(), reportingResource.getKey(), resourceLogs, responseStatus);
+                    logGeneratorState.getTransportStorage().store(logDefinition.getId(),
+                            reportingResourceByType.getKey(), resourceLogs, responseStatus);
                 }
-                log.debug(requestID + ": Complete payload for resource: " + reportingResource + " in log Definition" + logDefinition.getId() + ": " + resourceLogs);
+                log.debug(requestID + ": Complete payload for resource: " + reportingResourceByType.getKey() +
+                        " in log Definition" + logDefinition.getId() + ": " + resourceLogs);
                 resourceLogsList.clear();
             }
             currentPayloadCount++;
@@ -131,4 +145,18 @@ private List<Resource> getResourceSubsetByPostCount(String resourceName, int res
         return resourcesInResourceModel.subList(resourceStartIndex, resourceEndIndex)
                 .stream().map(GeneratorResource::getOTelResource).collect(Collectors.toList());
     }
+
+    private List<Resource> getFilteredResources(String resourceName, Map<String, String> filters) {
+        List<Resource> filteredResources = new ArrayList<>();
+        List<GeneratorResource> allResources = ResourceModelProvider.getResourceModel(requestID)
+                .get(resourceName).stream()
+                .filter(GeneratorResource::isActive)
+                .toList();
+        for (GeneratorResource eachResource: allResources) {
+            if (eachResource.getEvaluatedAttributes().entrySet().containsAll(filters.entrySet())) {
+                filteredResources.add(eachResource.getOTelResource());
+            }
+        }
+        return filteredResources;
+    }
 }
diff --git a/src/main/java/io/opentelemetry/contrib/generator/telemetry/logs/dto/LogDefinition.java b/src/main/java/io/opentelemetry/contrib/generator/telemetry/logs/dto/LogDefinition.java
@@ -33,13 +33,16 @@ public class LogDefinition {
 
     private String severityOrderFunction;
     private Map<String, Integer> reportingResourcesCounts;
+    private Map<String, Set<String>> filteredReportingResources;
     private Integer payloadFrequencySeconds;
     private Integer payloadCount;
     private Integer copyCount;
     private Set<String> copyResourceAttributes;
     private Map<String, Object> attributes;
     @JsonIgnore
     private String id;
+    @JsonIgnore
+    private Map<String, Map<String, String>> parsedFilteredReportingResources;
 
     public long validate(String requestID, Set<String> allResourceTypes, Integer globalPayloadFrequencySeconds, int logIndex) {
         id = "log_by_ttg_" + logIndex;
@@ -50,7 +53,8 @@ public long validate(String requestID, Set<String> allResourceTypes, Integer glo
             copyResourceAttributes = new HashSet<>();
         }
         validateMandatoryFields();
-        validateResourceTypesCount(allResourceTypes);
+        validateResourceTypes(allResourceTypes);
+        parseFilteredReportingResources();
         addRequestIDAndLogNameToValueFunction(requestID);
         attributes = GeneratorUtils.addArgsToAttributeExpressions(requestID, "log", id, attributes);
         return validatePayloadFrequency(globalPayloadFrequencySeconds);
@@ -60,8 +64,10 @@ private void validateMandatoryFields() {
         if (payloadCount == null || payloadCount < 1) {
             throw new GeneratorException("Payload count cannot be less than 1. Update the value in log " + this);
         }
-        if (MapUtils.emptyIfNull(reportingResourcesCounts).isEmpty()) {
-            throw new GeneratorException("Mandatory field 'reportingResourcesCount' not provided in log definition YAML for log " + this);
+        if (MapUtils.emptyIfNull(reportingResourcesCounts).isEmpty() &&
+                MapUtils.emptyIfNull(filteredReportingResources).isEmpty()) {
+            throw new GeneratorException("At least one resource type must be specified in either reportingResources" +
+                    " or filteredReportingResources for" + this);
         }
         if (StringUtils.defaultString(severityOrderFunction).isBlank()) {
             throw new GeneratorException("Mandatory field 'severityFrequency' not provided in log definition YAML for log " + this);
@@ -107,7 +113,13 @@ private void validateAttributes() {
         attributes = GeneratorUtils.validateAttributes(attributes);
     }
 
-    private void validateResourceTypesCount(Set<String> allResourceTypes) {
+    private void validateResourceTypes(Set<String> allResourceTypes) {
+        Set<String> resourcesInCounts = MapUtils.emptyIfNull(reportingResourcesCounts).keySet();
+        Set<String> resourcesInFilteredResources = MapUtils.emptyIfNull(filteredReportingResources).keySet();
+        if (!Collections.disjoint(resourcesInCounts, resourcesInFilteredResources)) {
+            throw new GeneratorException("'reportingResourcesCounts' and 'filteredReportingResources' cannot have " +
+                    "the same resource type for log " + this);
+        }
         Map<String, Integer> resourceCount = new HashMap<>();
         for (Map.Entry<String, Integer> eachResource : MapUtils.emptyIfNull(reportingResourcesCounts).entrySet()) {
             if (eachResource.getKey().trim().length() == 0) {
@@ -124,6 +136,32 @@ private void validateResourceTypesCount(Set<String> allResourceTypes) {
             else resourceCount.put(eachResource.getKey().trim(), eachResource.getValue());
         }
         reportingResourcesCounts = resourceCount;
+        for (String resourceType: MapUtils.emptyIfNull(filteredReportingResources).keySet()) {
+            if (!allResourceTypes.contains(resourceType)) {
+                throw new GeneratorException("Invalid resource type (" + resourceType + ") found in " +
+                        "'filteredReportingResources' for log " + this);
+            }
+        }
+    }
+
+    private void parseFilteredReportingResources() {
+        if (!MapUtils.emptyIfNull(filteredReportingResources).isEmpty()) {
+            parsedFilteredReportingResources = new HashMap<>();
+            for (Map.Entry<String, Set<String>> eachFilteredReportingResource: filteredReportingResources.entrySet()) {
+                String resourceType = eachFilteredReportingResource.getKey();
+                parsedFilteredReportingResources.put(resourceType, new HashMap<>());
+                for (String attributeFilter : eachFilteredReportingResource.getValue()) {
+                    String[] filterTokens = attributeFilter.split("=");
+                    if (filterTokens.length != 2) {
+                        log.warn("Attribute filter " + attributeFilter + " provided for resource type " +
+                                resourceType + " in the log definition YAML for log " + this + " is not valid. " +
+                                "Must contain a single '='");
+                        continue;
+                    }
+                    parsedFilteredReportingResources.get(resourceType).put(filterTokens[0], filterTokens[1]);
+                }
+            }
+        }
     }
 
     private void addRequestIDAndLogNameToValueFunction(String requestID) {
diff --git a/src/main/java/io/opentelemetry/contrib/generator/telemetry/metrics/dto/MetricDefinition.java b/src/main/java/io/opentelemetry/contrib/generator/telemetry/metrics/dto/MetricDefinition.java
@@ -16,6 +16,7 @@
 
 package io.opentelemetry.contrib.generator.telemetry.metrics.dto;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import io.opentelemetry.contrib.generator.core.exception.GeneratorException;
 import io.opentelemetry.contrib.generator.telemetry.misc.Constants;
 import io.opentelemetry.contrib.generator.telemetry.misc.GeneratorUtils;
@@ -46,6 +47,7 @@ public class MetricDefinition implements Cloneable {
     private Map<String, Set<String>> filteredReportingResources;
     private Set<String> copyResourceAttributes;
     private Map<String, Object> attributes;
+    @JsonIgnore
     private Map<String, Map<String, String>> parsedFilteredReportingResources;
 
     public void validate(String requestID, Set<String> allResourceTypes, Integer globalPayloadFrequency,
diff --git a/src/test/java/io/opentelemetry/contrib/generator/telemetry/TestAllGeneratorsWithJSONInput.java b/src/test/java/io/opentelemetry/contrib/generator/telemetry/TestAllGeneratorsWithJSONInput.java
@@ -55,11 +55,14 @@ public void validatePacketCounts() {
         int METRIC_REPORTING_RESOURCES_COUNT = NETWORK_INTERFACE_COUNT + CONTAINER_COUNT + MACHINE_COUNT + NODE_COUNT +
                 POD_COUNT + DISK_COUNT + AWS_EBS_COUNT + AWS_RDS_COUNT;
         int LOG_REPORTING_RESOURCES_COUNT = CONTAINER_COUNT + NODE_COUNT + 2 * POD_COUNT + MACHINE_COUNT;
+        int LOG_FILTERED_RESOURCES = 5;
+        int LOG_FILTERED_PAYLOADS = 20;
         int metricPayloadCount = 10;
         int logsPayloadCount = 20;
         int POD_EVENTS_COUNT = 30 * 5;
         int expectedMetricPackets = METRIC_REPORTING_RESOURCES_COUNT * metricPayloadCount;
-        int expectedLogsPackets = LOG_REPORTING_RESOURCES_COUNT * logsPayloadCount + POD_EVENTS_COUNT;
+        int expectedLogsPackets = LOG_REPORTING_RESOURCES_COUNT * logsPayloadCount + POD_EVENTS_COUNT +
+                (LOG_FILTERED_RESOURCES * LOG_FILTERED_PAYLOADS);
         int expectedSpanPackets = 11518;
         Assert.assertEquals(testStore.getMetricsPacketCount(), expectedMetricPackets, "Mismatch in expected metric packets count");
         Assert.assertEquals(testStore.getLogsPacketCount(), expectedLogsPackets, "Mismatch in expected log packets count");
@@ -70,7 +73,7 @@ public void validatePacketCounts() {
     public void validateStorageCounts() {
         Assert.assertEquals(transportStorage.getStoredMetricsPayloads().size(), 8,
                 "Mismatch in resource type counts for metric payloads");
-        Assert.assertEquals(transportStorage.getStoredLogsPayloads().size(), 4,
+        Assert.assertEquals(transportStorage.getStoredLogsPayloads().size(), 5,
                 "Mismatch in resource type counts for log payloads");
         Assert.assertEquals(transportStorage.getStoredLogsPayloads().get("log_by_ttg_0").size(), 3,
                 "Mismatch in resource type counts for log payloads");
@@ -80,7 +83,7 @@ public void validateStorageCounts() {
                 "Mismatch in resource type counts for trace payloads");
         Assert.assertEquals(transportStorage.getMetricsResponses().size(), 8,
                 "Mismatch in resource type counts for metric response statuses");
-        Assert.assertEquals(transportStorage.getLogsResponses().size(), 4,
+        Assert.assertEquals(transportStorage.getLogsResponses().size(), 5,
                 "Mismatch in resource type counts for log response statuses");
         Assert.assertEquals(transportStorage.getTracesResponses().size(), 8,
                 "Mismatch in resource type counts for trace response statuses");
diff --git a/src/test/java/io/opentelemetry/contrib/generator/telemetry/TestLogsGenerator.java b/src/test/java/io/opentelemetry/contrib/generator/telemetry/TestLogsGenerator.java
@@ -38,12 +38,14 @@ public class TestLogsGenerator {
     private final int CONTAINER_COUNT_LOG = 150;
     private final int MACHINE_COUNT = 4;
     private final int NODE_COUNT = 25;
+    private final int FILTERED_NODE_COUNT = 5;
     private final int POD_COUNT = 10;
     private final int POD_EVENT_COUNT = 30;
     private final int POST_COUNT_K8S_LOG = 10;
     private final int POST_COUNT_LOG_LOG_1 = 2;
     private final int POST_COUNT_LOG_LOG_2 = 5;
     private final int POST_COUNT_K8S_EVENT = 5;
+    private final int FILTERED_NODE_PAYLOAD_COUNT = 2;
 
     @BeforeClass
     public void generateData() {
@@ -57,11 +59,13 @@ public void generateData() {
     @Test
     public void testPayloadAndPacketCounts() {
         //Check payload count = Summation of all post counts per log definition
-        int expectedPayloadCount = 2 * POST_COUNT_K8S_LOG + POST_COUNT_LOG_LOG_1 + 2 * POST_COUNT_LOG_LOG_2 + POST_COUNT_K8S_EVENT;
+        int expectedPayloadCount = 2 * POST_COUNT_K8S_LOG + POST_COUNT_LOG_LOG_1 + 2 * POST_COUNT_LOG_LOG_2 +
+                POST_COUNT_K8S_EVENT + FILTERED_NODE_PAYLOAD_COUNT;
         Assert.assertEquals(testStore.getLogsPayloads().size(), expectedPayloadCount, "Mismatch in payload count");
         //Check packet count = Summation (payload count * number of resources) for every log
         int expectedPacketCount = (CONTAINER_COUNT_K8S + POD_COUNT) * POST_COUNT_K8S_LOG + NODE_COUNT * POST_COUNT_LOG_LOG_1
-                + (MACHINE_COUNT + CONTAINER_COUNT_LOG) * POST_COUNT_LOG_LOG_2 + POD_EVENT_COUNT * POST_COUNT_K8S_EVENT;
+                + (MACHINE_COUNT + CONTAINER_COUNT_LOG) * POST_COUNT_LOG_LOG_2 +
+                POD_EVENT_COUNT * POST_COUNT_K8S_EVENT + FILTERED_NODE_COUNT * FILTERED_NODE_PAYLOAD_COUNT;
         Assert.assertEquals(testStore.getLogsPacketCount(), expectedPacketCount, "Mismatch in resource logs packet count");
         //Check log count for each log = number of reporting resources * number of payloads defined per log definition
     }
@@ -73,7 +77,9 @@ public void testLogsCounts() {
         int log1_Count = NODE_COUNT * POST_COUNT_LOG_LOG_1;
         int log2_Count = (MACHINE_COUNT + CONTAINER_COUNT_LOG) * POST_COUNT_LOG_LOG_2;
         int k8sEventsCount = POD_EVENT_COUNT * POST_COUNT_K8S_EVENT;
-        Assert.assertEquals(testStore.getLogsPacketCount(), k8sLogs_Count + log1_Count + log2_Count + k8sEventsCount,
+        int filteredNodeEvents = FILTERED_NODE_COUNT * FILTERED_NODE_PAYLOAD_COUNT;
+        Assert.assertEquals(testStore.getLogsPacketCount(), k8sLogs_Count + log1_Count + log2_Count +
+                        k8sEventsCount + filteredNodeEvents,
                 "Mismatch in logs count");
     }
 }
diff --git a/src/test/resources/test-definitions/logs-test-combined.json b/src/test/resources/test-definitions/logs-test-combined.json
@@ -41,6 +41,13 @@
         "event.reason": "roundRobin([\"Pulling\", \"Pulled\", \"Created\", \"Pulling\", \"Backoff\"])",
         "event.message": "roundRobin([\"Pulling image k8s.gcr.io/echoserver:1.8\", \"Image downloaded\", \"Created container\", \"Pulling image cjknsjc/ccsdc:fff\", \"Error: ImagePullBackoff\"])"
       }
+    },
+    {
+      "payloadCount": 20,
+      "severityOrderFunction": "severityDistributionCount([\"INFO\", \"ERROR\", \"WARN\", \"DEBUG\"], [5, 1, 1, 4])",
+      "filteredReportingResources": {
+        "node": ["k8s.cluster.name=cluster-aukus"]
+      }
     }
   ]
 }
diff --git a/src/test/resources/test-definitions/logs-test.yaml b/src/test/resources/test-definitions/logs-test.yaml
@@ -51,3 +51,9 @@ logs:
       event.domain: 'roundRobin(["k8s"])'
       event.reason: 'roundRobin(["Pulling", "Pulled", "Created", "Pulling", "Backoff"])'
       event.message: 'roundRobin(["Pulling image k8s.gcr.io/echoserver:1.8", "Image downloaded", "Created container", "Pulling image cjknsjc/ccsdc:fff", "Error: ImagePullBackoff"])'
+  - severityOrderFunction: 'severityDistributionCount(["ERROR", "WARN", "DEBUG"], [1, 1, 4])'
+    payloadFrequencySeconds: 20
+    payloadCount: 2
+    copyCount: 10
+    filteredReportingResources:
+      node: ["k8s.cluster.name=cluster-aukus"]

Original file line number	Diff line number	Diff line change
`@@ -45,6 +45,15 @@`
`45`	`45`	`"event.reason": "roundRobin([\"Pulling\", \"Pulled\", \"Created\", \"Pulling\", \"Backoff\"])",`
`46`	`46`	`"event.message": "roundRobin([\"Pulling image k8s.gcr.io/echoserver:1.8\", \"Image downloaded\", \"Created container\", \"Pulling image cjknsjc/ccsdc:fff\", \"Error: ImagePullBackoff\"])"`
`47`	`47`	`}`
	`48`	`+ },`
	`49`	`+ {`
	`50`	`+ "severityOrderFunction": "severityDistributionCount([\"ERROR\", \"WARN\", \"DEBUG\"], [1, 1, 4])",`
	`51`	`+ "payloadFrequencySeconds": 20,`
	`52`	`+ "payloadCount": 2,`
	`53`	`+ "copyCount": 200,`
	`54`	`+ "filteredReportingResources": {`
	`55`	`+ "node": ["k8s.cluster.name=cluster-skyrim"]`
	`56`	`+ }`
`48`	`57`	`}`
`49`	`58`	`]`
`50`	`59`	`}`
Original file line number	Diff line number	Diff line change
`@@ -41,6 +41,13 @@`
`41`	`41`	`"event.reason": "roundRobin([\"Pulling\", \"Pulled\", \"Created\", \"Pulling\", \"Backoff\"])",`
`42`	`42`	`"event.message": "roundRobin([\"Pulling image k8s.gcr.io/echoserver:1.8\", \"Image downloaded\", \"Created container\", \"Pulling image cjknsjc/ccsdc:fff\", \"Error: ImagePullBackoff\"])"`
`43`	`43`	`}`
	`44`	`+ },`
	`45`	`+ {`
	`46`	`+ "payloadCount": 20,`
	`47`	`+ "severityOrderFunction": "severityDistributionCount([\"INFO\", \"ERROR\", \"WARN\", \"DEBUG\"], [5, 1, 1, 4])",`
	`48`	`+ "filteredReportingResources": {`
	`49`	`+ "node": ["k8s.cluster.name=cluster-aukus"]`
	`50`	`+ }`
`44`	`51`	`}`
`45`	`52`	`]`
`46`	`53`	`}`