feat(logging): log stream provider to cache log stream

Author: Nika Hassani

packages/logging_cloudwatch/aws_logging_cloudwatch/lib/src/cloudwatch_logger_plugin.dart

@@ -32,7 +32,7 @@ import 'package:meta/meta.dart';
 // The maximum number of log events in a batch is 10,000.
 
 const int _maxNumberOfLogEventsInBatch = 10000;
-const int _maxLogEventsTimeSpanInBatch = 24 * 3600;
+const int _maxLogEventsTimeSpanInBatch = 24 * 3600 * 1000;
 const int _maxLogEventsBatchSize = 1048576;
 const int _baseBufferSize = 26;
 const int _maxLogEventSize = 256000;
@@ -138,51 +138,26 @@ class CloudWatchLoggerPlugin extends AWSLoggerPlugin
 
   Future<void> _startSyncingIfNotInProgress() async {
     Future<void> startSyncing() async {
-      String logStream;
-      try {
-        logStream = await _logStreamProvider.logStream;
-      } on Exception catch (e) {
-        if (await _logStore.isFull(_pluginConfig.localStoreMaxSizeInMB)) {
-          await _logStore.clear();
-          logger.warn(
-            'Reached local store max size of: '
-            '${_pluginConfig.localStoreMaxSizeInMB}.Hence logs are deleted from '
-            'local store.',
-          );
-        }
-        logger.error('Failed to create CloudWatch log stream', e);
-        return;
-      }
-
       final batcheStream = _getLogBatchesToSync();
       await for (final (logs, events) in batcheStream) {
-        try {
-          final response = await _sendToCloudWatch(events, logStream);
-          if (response.rejectedLogEventsInfo?.tooNewLogEventStartIndex !=
-              null) {
-            break;
-          }
-          await _logStore.deleteItems(logs);
-        } on Exception catch (e) {
-          if (await _logStore.isFull(_pluginConfig.localStoreMaxSizeInMB)) {
-            await _logStore.deleteItems(logs);
-            logger.warn(
-              'Reached local store max size of: '
-              '${_pluginConfig.localStoreMaxSizeInMB}.Hence logs are deleted '
-              'from local store.',
-            );
-          }
-          logger.error('Failed to sync batched logs to CloudWatch', e);
+        final response = await _sendToCloudWatch(events);
+        if (response.rejectedLogEventsInfo?.tooNewLogEventStartIndex != null) {
+          throw _TooNewLogEventException(
+            'logs can not be more than 2 hours in the future',
+          );
         }
+        await _logStore.deleteItems(logs);
       }
     }
 
     if (!_syncing) {
+      // TODO(nikahsn): disable log rotation.
       _syncing = true;
       try {
         await startSyncing();
       } on Exception catch (e) {
         logger.error('Failed to sync logs to CloudWatch.', e);
+        // TODO(nikahsn): enable log rotation if the log store is full
       } finally {
         _syncing = false;
       }
@@ -200,43 +175,49 @@ class CloudWatchLoggerPlugin extends AWSLoggerPlugin
 
   Future<PutLogEventsResponse> _sendToCloudWatch(
     List<InputLogEvent> logEvents,
-    String logStreamName,
   ) async {
+    final logStreamName = await _logStreamProvider.logStream;
     final request = PutLogEventsRequest(
       logGroupName: _pluginConfig.logGroupName,
       logStreamName: logStreamName,
       logEvents: logEvents,
     );
-    return _client.putLogEvents(request).result;
+    try {
+      return _client.putLogEvents(request).result;
+    } on ResourceNotFoundException {
+      await _logStreamProvider.createLogStream(logStreamName);
+      return _client.putLogEvents(request).result;
+    } on Exception {
+      rethrow;
+    }
   }
 
   Stream<_LogBatch> _getLogBatchesToSync() async* {
     final queuedLogs = (await _logStore.getAll()).toList();
-    if (queuedLogs.isEmpty) {
-      yield ([], []);
-    }
-    final logEvents = <InputLogEvent>[];
-    final logQueues = <QueuedItem>[];
-    var totalByteSize = 0;
-
-    for (final currentLog in queuedLogs) {
-      final currentLogEvent = currentLog.toInputLogEvent();
-      final size = currentLogEvent.message.length + _baseBufferSize;
-      if (totalByteSize + size >= _maxLogEventsBatchSize ||
-          logEvents.length >= _maxNumberOfLogEventsInBatch ||
-          (logEvents.length > 1 &&
-              currentLogEvent.timestamp - logEvents.first.timestamp >=
-                  _maxLogEventsTimeSpanInBatch)) {
-        yield (logQueues, logEvents);
-        totalByteSize = 0;
-        logEvents.clear();
-        logQueues.clear();
+    if (queuedLogs.isNotEmpty) {
+      final logEvents = <InputLogEvent>[];
+      final logQueues = <QueuedItem>[];
+      var totalByteSize = 0;
+
+      for (final currentLog in queuedLogs) {
+        final currentLogEvent = currentLog.toInputLogEvent();
+        final size = currentLogEvent.message.length + _baseBufferSize;
+        if (totalByteSize + size >= _maxLogEventsBatchSize ||
+            logEvents.length >= _maxNumberOfLogEventsInBatch ||
+            (logEvents.length > 1 &&
+                currentLogEvent.timestamp - logEvents.first.timestamp >=
+                    _maxLogEventsTimeSpanInBatch)) {
+          yield (logQueues, logEvents);
+          totalByteSize = 0;
+          logEvents.clear();
+          logQueues.clear();
+        }
+        totalByteSize += size;
+        logEvents.add(currentLogEvent);
+        logQueues.add(currentLog);
       }
-      totalByteSize += size;
-      logEvents.add(currentLogEvent);
-      logQueues.add(currentLog);
+      yield (logQueues, logEvents);
     }
-    yield (logQueues, logEvents);
   }
 
   /// Whether a [logEntry] should be logged by this plugin.
@@ -314,3 +295,8 @@ extension on LogEntry {
     );
   }
 }
+
+class _TooNewLogEventException implements Exception {
+  _TooNewLogEventException(this.message);
+  final String message;
+}

packages/logging_cloudwatch/aws_logging_cloudwatch/lib/src/log_stream_provider.dart

@@ -1,6 +1,8 @@
 // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
 
+import 'dart:async';
+
 import 'package:aws_logging_cloudwatch/src/sdk/cloud_watch_logs.dart';
 import 'package:intl/intl.dart';
 
@@ -9,10 +11,12 @@ import 'package:intl/intl.dart';
 /// [CloudWatchLogStreamProvider]
 /// {@endtemplate}
 abstract class CloudWatchLogStreamProvider {
-  /// Returns CloudWatch log stream name to use for sending logs to CloudWatch.
-  ///
-  /// It creates the log stream if it does not exists.
-  Future<String> get logStream;
+  /// Returns log stream name from cache. if cache is missing it
+  /// calls [createLogStream] and return the log stream name.
+  FutureOr<String> get logStream;
+
+  /// Creates the log stream and add it to the cache.
+  Future<void> createLogStream(String logStreamName);
 }
 
 /// {@template aws_logging_cloudwatch.default_cloudwatch_logstream_provider}
@@ -36,15 +40,22 @@ class DefaultCloudWatchLogStreamProvider
   final String _logGroupName;
   final CloudWatchLogsClient _client;
   static final DateFormat _dateFormat = DateFormat('yyyy-MM-dd');
+  final _cachedLogStreams = <String>{};
 
-  /// Creates CloudWatch log stream if does not exists and returns
-  /// the log stream name.
-  ///
-  /// Throws an [Exception] if fails to create the log stream.
   @override
-  Future<String> get logStream async {
+  FutureOr<String> get logStream async {
     final logStreamName =
         _logStreamName ?? _dateFormat.format(DateTime.timestamp());
+    if (_cachedLogStreams.contains(logStreamName)) {
+      return logStreamName;
+    }
+    await createLogStream(logStreamName);
+    _cachedLogStreams.add(logStreamName);
+    return logStreamName;
+  }
+
+  @override
+  Future<void> createLogStream(String logStreamName) async {
     try {
       await _client
           .createLogStream(
@@ -55,10 +66,9 @@ class DefaultCloudWatchLogStreamProvider
           )
           .result;
     } on ResourceAlreadyExistsException {
-      return logStreamName;
+      return;
     } on Exception {
       rethrow;
     }
-    return logStreamName;
   }
 }

packages/logging_cloudwatch/aws_logging_cloudwatch/lib/src/stoppable_timer.dart

@@ -16,6 +16,7 @@ class StoppableTimer {
     required Future<void> Function() callback,
     required void Function(Object) onError,
   })  : _callback = callback,
+        _onError = onError,
         _timer = Timer.periodic(duration, (Timer t) {
           callback().catchError((Object e) {
             onError(e);
@@ -25,12 +26,15 @@ class StoppableTimer {
 
   /// [Duration] between invocations of the provided callback function.
   final Duration duration;
-  final void Function() _callback;
+  final Future<void> Function() _callback;
+  final void Function(Object) _onError;
 
   /// Start the timer.
   void start() {
     if (_timer.isActive) return;
-    _timer = Timer.periodic(duration, (Timer t) => _callback());
+    _timer = Timer.periodic(duration, (Timer t) {
+      _callback().catchError(_onError);
+    });
   }
 
   /// Stop the timer.