feat: add fluentbit monitoring

README.md

@@ -94,6 +94,7 @@ You can browse the documentation at https://constructs.dev/packages/cdk-monitori
 | AWS SQS Queue (`.monitorSqsQueue()`, `.monitorSqsQueueWithDlq()`) | Message count, age, size | Message count, age, DLQ incoming messages | |
 | AWS Step Functions (`.monitorStepFunction()`, `.monitorStepFunctionActivity()`, `monitorStepFunctionLambdaIntegration()`, `.monitorStepFunctionServiceIntegration()`) | Execution count and breakdown per state | Duration, failed, failed rate, aborted, throttled, timed out executions | |
 | AWS Web Application Firewall (`.monitorWebApplicationFirewallAclV2()`) | Allowed/blocked requests | Blocked requests count/rate | |
+| FluentBit (`.monitorFluentBit()`) | Num of input records, Output failures & retries, Filter metrics, Storage metrics | | FluentBit needs proper configuration with  metrics enabled: [Official sample configuration](https://github.com/aws-samples/amazon-ecs-firelens-examples/tree/mainline/examples/fluent-bit/send-fb-internal-metrics-to-cw). This function creates MetricFilters to publish all FluentBit metrics. |
 | Custom metrics (`.monitorCustom()`) | Addition of custom metrics into the dashboard (each group is a widget) | | Supports anomaly detection |
 
 

lib/facade/MonitoringFacade.ts

@@ -70,6 +70,8 @@ import {
   FargateNetworkLoadBalancerMonitoringProps,
   FargateServiceMonitoring,
   FargateServiceMonitoringProps,
+  FluentBitMonitoring,
+  FluentBitMonitoringProps,
   getQueueProcessingEc2ServiceMonitoring,
   getQueueProcessingFargateServiceMonitoring,
   GlueJobMonitoring,
@@ -753,4 +755,10 @@ export class MonitoringFacade extends MonitoringScope {
     this.addSegment(segment, props);
     return this;
   }
+
+  monitorFluentBit(props: FluentBitMonitoringProps) {
+    const segment = new FluentBitMonitoring(this, props);
+    this.addSegment(segment, props);
+    return this;
+  }
 }

lib/monitoring/fluentbit/FluentBitConstants.ts

@@ -0,0 +1,29 @@
+export enum FluentBitStorageMetricTag {
+  TOTAL_CHUNKS = "total_chunks",
+  MEM_CHUNKS = "mem_chunks",
+  FS_CHUNKS = "fs_chunks",
+  FS_CHUNKS_UP = "fs_chunks_up",
+  FS_CHUNKS_DOWN = "fs_chunks_down",
+}
+
+export enum FluentBitOutputMetricTag {
+  OUTPUT_RETRIES = "fluentbit_output_retries_total",
+  OUTPUT_RETRIES_FAILED = "fluentbit_output_retries_failed_total",
+  OUTPUT_ERRORS = "fluentbit_output_errors_total",
+  OUTPUT_DROPPED_RECORDS = "fluentbit_output_dropped_records_total",
+}
+
+export enum FluentBitInputMetricTag {
+  INPUT_RECORDS = "fluentbit_input_records_total",
+}
+export enum FluentBitFilterMetricTag {
+  FILTER_EMIT_RECORDS = "fluentbit_filter_emit_records_total",
+  FILTER_DROP_RECORDS = "fluentbit_filter_drop_records_total",
+  FILTER_ADD_RECORDS = "fluentbit_filter_add_records_total",
+}
+
+export enum FluentBitMetricsWithoutWidget {
+  INPUT_BYTES = "fluentbit_input_bytes_total",
+  OUTPUT_PROC_RECORDS = "fluentbit_output_proc_records_total",
+  OUTPUT_PROC_BYTES = "fluentbit_output_proc_bytes_total",
+}

lib/monitoring/fluentbit/FluentBitMetricFactory.ts

@@ -0,0 +1,93 @@
+import { FilterPattern, ILogGroup, MetricFilter } from "aws-cdk-lib/aws-logs";
+import {
+  FluentBitFilterMetricTag,
+  FluentBitInputMetricTag,
+  FluentBitMetricsWithoutWidget,
+  FluentBitOutputMetricTag,
+  FluentBitStorageMetricTag,
+} from "./FluentBitConstants";
+import { MetricFactory, MetricStatistic, MonitoringScope } from "../../common";
+
+export interface FluentBitMetricFactoryProps {
+  /**
+   * Namespace that metrics will be emitted to.
+   * @default metric factory default
+   */
+  readonly namespace?: string;
+}
+
+export class FluentBitMetricFactory {
+  protected readonly metricFactory: MetricFactory;
+  protected readonly namespace: string;
+  protected readonly scope: MonitoringScope;
+
+  constructor(scope: MonitoringScope, props: FluentBitMetricFactoryProps) {
+    this.scope = scope;
+    this.metricFactory = scope.createMetricFactory();
+    this.namespace =
+      props.namespace ??
+      this.metricFactory.getNamespaceWithFallback(props.namespace);
+  }
+
+  filterMetrics(logGroup: ILogGroup) {
+    return Object.values(FluentBitFilterMetricTag).map((metricName) => {
+      return this.pluginMetric(logGroup, metricName);
+    });
+  }
+
+  outputMetrics(logGroup: ILogGroup) {
+    return Object.values(FluentBitOutputMetricTag).map((metricName) => {
+      return this.pluginMetric(logGroup, metricName);
+    });
+  }
+
+  inputMetrics(logGroup: ILogGroup) {
+    return Object.values(FluentBitInputMetricTag).map((metricName) => {
+      return this.pluginMetric(logGroup, metricName);
+    });
+  }
+
+  private pluginMetric(logGroup: ILogGroup, metricName: string) {
+    const metricFilter = new MetricFilter(
+      this.scope,
+      `FluentBit-${metricName}-${logGroup}-MetricFilter`,
+      {
+        logGroup: logGroup,
+        filterPattern: FilterPattern.literal(`{ $.metric = "${metricName}" }`),
+        metricNamespace: this.namespace,
+        metricName,
+        metricValue: "$.value",
+      }
+    );
+    return metricFilter.metric({
+      statistic: MetricStatistic.MAX,
+    });
+  }
+
+  storageMetrics(logGroup: ILogGroup) {
+    return Object.values(FluentBitStorageMetricTag).map((metricName) => {
+      const valueString = `$.storage_layer.chunks.${metricName}`;
+      const metricFilter = new MetricFilter(
+        this.scope,
+        `FluentBit-${metricName}-${logGroup}-MetricFilter`,
+        {
+          logGroup: logGroup,
+          filterPattern: FilterPattern.literal(`{ ${valueString} = * }`),
+          metricNamespace: this.namespace,
+          metricName,
+          metricValue: `${valueString}`,
+        }
+      );
+      const metric = metricFilter.metric({
+        statistic: MetricStatistic.MAX,
+      });
+      return metric;
+    });
+  }
+
+  metricsWithoutWidgets(logGroup: ILogGroup) {
+    Object.values(FluentBitMetricsWithoutWidget).forEach((metricName) =>
+      this.pluginMetric(logGroup, metricName)
+    );
+  }
+}

lib/monitoring/fluentbit/FluentBitMonitoring.ts

@@ -0,0 +1,119 @@
+import { GraphWidget, IWidget, Metric } from "aws-cdk-lib/aws-cloudwatch";
+import { ILogGroup } from "aws-cdk-lib/aws-logs";
+import {
+  FluentBitMetricFactory,
+  FluentBitMetricFactoryProps,
+} from "./FluentBitMetricFactory";
+import {
+  BaseMonitoringProps,
+  CountAxisFromZero,
+  DefaultGraphWidgetHeight,
+  HalfWidth,
+  MetricWithAlarmSupport,
+  Monitoring,
+  MonitoringScope,
+} from "../../common";
+import { MonitoringHeaderWidget } from "../../dashboard";
+
+export interface FluentBitMonitoringProps
+  extends FluentBitMetricFactoryProps,
+    BaseMonitoringProps {
+  /**
+   * Log group that contains FluentBit metric logs
+   */
+  readonly logGroup: ILogGroup;
+
+  /**
+   * Metrics for input bytes total, output bytes total and output records total are not shown on default dashboard.
+   * If you want to get MetricFilters created to have those metrics present in CloudWatch set this flag to true
+   * @default false
+   */
+  readonly createOptionalMetricFilters?: boolean;
+}
+
+export class FluentBitMonitoring extends Monitoring {
+  protected readonly logGroupName: string;
+  protected readonly metricFactory: FluentBitMetricFactory;
+  protected readonly storageMetrics: Metric[];
+  protected readonly inputMetrics: Metric[];
+  protected readonly outputMetrics: Metric[];
+  protected readonly filterMetrics: Metric[];
+
+  constructor(scope: MonitoringScope, props: FluentBitMonitoringProps) {
+    super(scope, props);
+    this.logGroupName = props.logGroup.logGroupName;
+    this.metricFactory = new FluentBitMetricFactory(scope, props);
+
+    this.storageMetrics = this.metricFactory.storageMetrics(props.logGroup);
+    this.inputMetrics = this.metricFactory.inputMetrics(props.logGroup);
+    this.outputMetrics = this.metricFactory.outputMetrics(props.logGroup);
+    this.filterMetrics = this.metricFactory.filterMetrics(props.logGroup);
+    if (props.createOptionalMetricFilters) {
+      this.metricFactory.metricsWithoutWidgets(props.logGroup);
+    }
+  }
+
+  widgets(): IWidget[] {
+    return [
+      this.createTitleWidget(),
+      this.inputMetricsWidget(),
+      this.outputMetricsWidget(),
+      this.filterMetricsWidget(),
+      this.storageMetricsWidget(),
+    ];
+  }
+
+  summaryWidgets(): IWidget[] {
+    return [
+      this.createTitleWidget(),
+      this.outputMetricsWidget(),
+      this.storageMetricsWidget(),
+    ];
+  }
+
+  private createTitleWidget() {
+    return new MonitoringHeaderWidget({
+      title: "FluentBit",
+    });
+  }
+  private inputMetricsWidget() {
+    return this.createMetricWidget(
+      [...Object.values(this.inputMetrics)],
+      "Input Metrics"
+    );
+  }
+  private outputMetricsWidget() {
+    return this.createMetricWidget(
+      [...Object.values(this.outputMetrics)],
+      "Output Metrics"
+    );
+  }
+
+  private filterMetricsWidget() {
+    return this.createMetricWidget(
+      [...Object.values(this.filterMetrics)],
+      "Filter Metrics"
+    );
+  }
+
+  private storageMetricsWidget() {
+    return this.createMetricWidget(
+      [...Object.values(this.storageMetrics)],
+      "Storage Metrics"
+    );
+  }
+
+  private createMetricWidget(
+    metrics: MetricWithAlarmSupport[],
+    title: string
+  ): GraphWidget {
+    return new GraphWidget({
+      width: HalfWidth,
+      height: DefaultGraphWidgetHeight,
+      title,
+      left: metrics,
+      leftAnnotations: undefined,
+      leftYAxis: CountAxisFromZero,
+    });
+  }
+}

lib/monitoring/fluentbit/index.ts

@@ -0,0 +1,3 @@
+export * from "./FluentBitConstants";
+export * from "./FluentBitMetricFactory";
+export * from "./FluentBitMonitoring";

lib/monitoring/index.ts

@@ -27,3 +27,4 @@ export * from "./aws-step-functions";
 export * from "./aws-synthetics";
 export * from "./aws-wafv2";
 export * from "./custom";
+export * from "./fluentbit";

test/monitoring/fluentbit/FluentBitMonitoring.test.ts

@@ -0,0 +1,33 @@
+import { Stack } from "aws-cdk-lib";
+import { Template } from "aws-cdk-lib/assertions";
+import { LogGroup } from "aws-cdk-lib/aws-logs";
+import { FluentBitMonitoring } from "../../../lib";
+import { addMonitoringDashboardsToStack } from "../../utils/SnapshotUtil";
+import { TestMonitoringScope } from "../TestMonitoringScope";
+
+test("snapshot test without all filters", () => {
+  const stack = new Stack();
+  const scope = new TestMonitoringScope(stack, "Scope");
+  const logGroup = new LogGroup(stack, "DummyLogGroup");
+  const monitoring = new FluentBitMonitoring(scope, {
+    logGroup,
+    namespace: "DummyNamespace",
+  });
+
+  addMonitoringDashboardsToStack(stack, monitoring);
+  expect(Template.fromStack(stack)).toMatchSnapshot();
+});
+
+test("snapshot test with all filters", () => {
+  const stack = new Stack();
+  const scope = new TestMonitoringScope(stack, "Scope");
+  const logGroup = new LogGroup(stack, "DummyLogGroup");
+  const monitoring = new FluentBitMonitoring(scope, {
+    logGroup,
+    namespace: "DummyNamespace",
+    createOptionalMetricFilters: true,
+  });
+
+  addMonitoringDashboardsToStack(stack, monitoring);
+  expect(Template.fromStack(stack)).toMatchSnapshot();
+});