feat: add fluentbit monitoring

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,105 @@
+import { Metric } from "aws-cdk-lib/aws-cloudwatch";
+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);
+  }
+
+  fluentBitFilterMetrics(logGroup: ILogGroup) {
+    const filterMetrics: Metric[] = [];
+    Object.values(FluentBitFilterMetricTag).forEach((metricName) => {
+      const metric = this.pluginMetric(logGroup, metricName);
+      filterMetrics.push(metric);
+    });
+    return filterMetrics;
+  }
+
+  fluentBitOutputMetrics(logGroup: ILogGroup) {
+    const outputMetrics: Metric[] = [];
+    Object.values(FluentBitOutputMetricTag).forEach((metricName) => {
+      const metric = this.pluginMetric(logGroup, metricName);
+      outputMetrics.push(metric);
+    });
+    return outputMetrics;
+  }
+
+  fluentBitInputMetrics(logGroup: ILogGroup) {
+    const inputMetrics: Metric[] = [];
+    Object.values(FluentBitInputMetricTag).forEach((metricName) => {
+      const metric = this.pluginMetric(logGroup, metricName);
+      inputMetrics.push(metric);
+    });
+    return inputMetrics;
+  }
+
+  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,
+    });
+  }
+
+  fluentBitStorageMetrics(logGroup: ILogGroup) {
+    const storageMetrics: Metric[] = [];
+    Object.values(FluentBitStorageMetricTag).forEach((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,
+      });
+      storageMetrics.push(metric);
+    });
+    return storageMetrics;
+  }
+
+  fluentBitMetricsWithoutWidgets(logGroup: ILogGroup) {
+    Object.values(FluentBitMetricsWithoutWidget).forEach((metricName) =>
+      this.pluginMetric(logGroup, metricName)
+    );
+  }
+}

lib/monitoring/fluentbit/FluentBitMonitoring.ts

@@ -0,0 +1,108 @@
+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 to which FluentBit emits metrics logs
+    */
+  readonly logGroup: ILogGroup;
+}
+
+export class FluentBitMonitoring extends Monitoring {
+  protected readonly logGroupName: string;
+  protected readonly metricFactory: FluentBitMetricFactory;
+  protected readonly fluentBitStorageMetrics: Metric[];
+  protected readonly fluentBitInputMetrics: Metric[];
+  protected readonly fluentBitOutputMetrics: Metric[];
+  protected readonly fluentBitFilterMetrics: Metric[];
+
+  constructor(scope: MonitoringScope, props: FluentBitMonitoringProps) {
+    super(scope, props);
+    this.logGroupName = props.logGroup.logGroupName;
+    this.metricFactory = new FluentBitMetricFactory(scope, props);
+
+    this.fluentBitStorageMetrics = this.metricFactory.fluentBitStorageMetrics(
+      props.logGroup
+    );
+    this.fluentBitInputMetrics = this.metricFactory.fluentBitInputMetrics(
+      props.logGroup
+    );
+    this.fluentBitOutputMetrics = this.metricFactory.fluentBitOutputMetrics(
+      props.logGroup
+    );
+    this.fluentBitFilterMetrics = this.metricFactory.fluentBitFilterMetrics(
+      props.logGroup
+    );
+    this.metricFactory.fluentBitMetricsWithoutWidgets(props.logGroup);
+  }
+
+  widgets(): IWidget[] {
+    return [
+      new MonitoringHeaderWidget({
+        title: "FluentBit",
+      }),
+      this.createFluentBitInputWidget(),
+      this.createFluentBitOutputWidget(),
+      this.createFluentBitFilterWidget(),
+      this.createFluentBitStorageWidget(),
+    ];
+  }
+
+  private createFluentBitInputWidget() {
+    return this.createFluentBitMetricWidget(
+      [...Object.values(this.fluentBitInputMetrics)],
+      "Input Metrics"
+    );
+  }
+  private createFluentBitOutputWidget() {
+    return this.createFluentBitMetricWidget(
+      [...Object.values(this.fluentBitOutputMetrics)],
+      "Output Metrics"
+    );
+  }
+
+  private createFluentBitFilterWidget() {
+    return this.createFluentBitMetricWidget(
+      [...Object.values(this.fluentBitFilterMetrics)],
+      "Filter Metrics"
+    );
+  }
+
+  private createFluentBitStorageWidget() {
+    return this.createFluentBitMetricWidget(
+      [...Object.values(this.fluentBitStorageMetrics)],
+      "Storage Metrics"
+    );
+  }
+
+  private createFluentBitMetricWidget(
+    metrics: MetricWithAlarmSupport[],
+    title: string
+  ): GraphWidget {
+    return new GraphWidget({
+      width: HalfWidth,
+      height: DefaultGraphWidgetHeight,
+      title: `${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,19 @@
+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", () => {
+  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();
+});