Cloudwatch integration demo

CMake/Dependencies/libawscpp-CMakeLists.txt

@@ -2,13 +2,28 @@ cmake_minimum_required(VERSION 3.6.3)
 project(libawscpp-download NONE)
 include(ExternalProject)
 
+if(CMAKE_SYSTEM_NAME STREQUAL "Linux")
+    message(STATUS "Configuring for Linux")
+    set(TARGET_ARCH LINUX)
+elseif(CMAKE_SYSTEM_NAME STREQUAL "Darwin")
+    message(STATUS "Configuring for macOS")
+    set(TARGET_ARCH APPLE)
+elseif(CMAKE_SYSTEM_NAME STREQUAL "Windows")
+    message(STATUS "Configuring for Windows")
+    set(TARGET_ARCH WINDOWS)
+endif()
+
 ExternalProject_Add(libawscpp-download
     GIT_REPOSITORY    https://github.com/aws/aws-sdk-cpp.git
     GIT_TAG           1.11.217
     LIST_SEPARATOR    "|"
-    CMAKE_ARGS       -DBUILD_SHARED_LIBS=OFF
-                     -DBUILD_ONLY=kinesisvideo|kinesis-video-webrtc-storage
+    CMAKE_ARGS       -DBUILD_SHARED_LIBS=ON
+                     -DENABLE_TESTING=OFF
+                     -DBUILD_ONLY=${BUILD_ONLY}
                      -DCMAKE_INSTALL_PREFIX=${OPEN_SRC_INSTALL_PREFIX}
+                     -DCMAKE_PREFIX_PATH=${OPEN_SRC_INSTALL_PREFIX}
+                     -DTARGET_ARCH=${TARGET_ARCH}
+                     -DCUSTOM_MEMORY_MANAGEMENT=OFF
     BUILD_ALWAYS      TRUE
     TEST_COMMAND      ""
 )

CMakeLists.txt

@@ -21,6 +21,7 @@ option(ENABLE_DATA_CHANNEL "Enable support for data channel" ON)
 option(ENABLE_KVS_THREADPOOL "Enable support for KVS thread pool in signaling" ON)
 option(INSTRUMENTED_ALLOCATORS "Enable memory instrumentation" OFF)
 option(ENABLE_AWS_SDK_IN_TESTS "Enable support for compiling AWS SDKs for tests" ON)
+option(ENABLE_AWS_SDK_INTEG "Enable building samples with cloudwatch" OFF)
 
 # Developer Flags
 option(BUILD_TEST "Build the testing tree." OFF)
@@ -233,9 +234,12 @@ if(BUILD_DEPENDENCIES)
 
   if(BUILD_TEST)
     build_dependency(gtest)
-
+    set(BUILD_ARGS
+            -DCMAKE_BUILD_TYPE=${CMAKE_BUILD_TYPE}
+            -DCMAKE_C_FLAGS=${CMAKE_C_FLAGS}
+            -DBUILD_ONLY=kinesisvideo|kinesis-video-webrtc-storage)
     if(ENABLE_AWS_SDK_IN_TESTS)
-      build_dependency(awscpp)
+      build_dependency(awscpp ${BUILD_ARGS})
     endif()
 
   endif()
@@ -247,9 +251,18 @@ if(BUILD_DEPENDENCIES)
   if (LINK_PROFILER)
     build_dependency(gperftools)
   endif()
+
   message(STATUS "Finished building dependencies.")
 endif()
 
+if(ENABLE_AWS_SDK_INTEG)
+  set(BUILD_ARGS
+          -DCMAKE_BUILD_TYPE=${CMAKE_BUILD_TYPE}
+          -DCMAKE_C_FLAGS=${CMAKE_C_FLAGS}
+          -DBUILD_ONLY=monitoring|logs)
+  build_dependency(awscpp ${BUILD_ARGS})
+endif()
+
 # building kvsCommonLws also builds kvspic
 set(BUILD_ARGS
         -DCMAKE_BUILD_TYPE=${CMAKE_BUILD_TYPE}
@@ -474,6 +487,9 @@ if (BUILD_SAMPLE)
   add_subdirectory(samples)
 endif()
 
+if(ENABLE_AWS_SDK_INTEG)
+  add_subdirectory(cloudwatch-integ)
+endif()
 if(BUILD_TEST)
   # adding ZLIB because aws sdk static link seems to be broken when zlib is needed
   if(NOT WIN32)

README.md

@@ -450,7 +450,7 @@ The certificate generating function ([createCertificateAndKey](https://awslabs.g
 **Important Note: It is recommended to rotate the certificates often - preferably for every peer connection to avoid a compromised client weakening the security of the new connections.**
 
 Take `kvsWebRTCClientMaster` as sample, add `RtcCertificate certificates[CERT_COUNT];` to **SampleConfiguration** in [Samples.h](./samples/Samples.h).
-Then pass in the pre-generated certificate in initializePeerConnection() in [Common.c](./samples/Common.c).
+Then pass in the pre-generated certificate in initializePeerConnection() in [Common.c](samples/lib/Common.c).
 
 ```c
 configuration.certificates[0].pCertificate = pSampleConfiguration->certificates[0].pCertificate;

cloudwatch-integ/CMakeLists.txt

@@ -0,0 +1,56 @@
+cmake_minimum_required(VERSION 3.6.3)
+
+project(KinesisVideoWebRTCClientSamplesCloudwatch LANGUAGES C CXX)
+set(CMAKE_CXX_STANDARD 11)
+
+message("OPEN_SRC_INSTALL_PREFIX=${OPEN_SRC_INSTALL_PREFIX}")
+
+include_directories(${OPEN_SRC_INSTALL_PREFIX}/include)
+include_directories(${OPEN_SRC_INCLUDE_DIRS})
+link_directories(${OPEN_SRC_INSTALL_PREFIX}/lib)
+
+find_package(ZLIB REQUIRED)
+find_package(AWSSDK REQUIRED COMPONENTS monitoring logs)
+
+set(CW_CONFIG_HEADER "configs/lr_static_h265_openssl.h" CACHE FILEPATH "Config header for the cloudwatch integrated demo")
+
+add_definitions(-DCW_CONFIG_HEADER="${CW_CONFIG_HEADER}")
+message(STATUS "Config header set to ${CW_CONFIG_HEADER}")
+
+add_executable(
+        kvsWebrtcClientMasterCW
+        Cloudwatch.cpp
+        CloudwatchLogs.cpp
+        CloudwatchMonitoring.cpp
+        ../samples/lib/Common.c
+        ../samples/lib/Utility.c
+        ../samples/lib/MetricsHandling.c
+        ../samples/lib/DataChannelHandling.c
+        ../samples/lib/SignalingMsgHandler.c
+        ../samples/lib/Media.c
+        kvsWebRTCClientMasterCloudwatch.cpp)
+target_link_libraries(kvsWebrtcClientMasterCW
+                      kvsWebrtcClient
+                      kvsWebrtcSignalingClient
+                      ${EXTRA_DEPS}
+                      kvsCommonLws kvspicUtils websockets kvssdp kvsstun
+                      ${AWSSDK_LINK_LIBRARIES})
+
+add_executable(
+        kvsWebrtcClientViewerCW
+        ../samples/lib/Common.c
+        ../samples/lib/Utility.c
+        ../samples/lib/MetricsHandling.c
+        ../samples/lib/DataChannelHandling.c
+        ../samples/lib/SignalingMsgHandler.c
+        ../samples/lib/Media.c
+        Cloudwatch.cpp
+        CloudwatchLogs.cpp
+        CloudwatchMonitoring.cpp
+        kvsWebRTCClientViewerCloudwatch.cpp)
+target_link_libraries(kvsWebrtcClientViewerCW
+        kvsWebrtcClient
+        kvsWebrtcSignalingClient
+        ${EXTRA_DEPS}
+        kvsCommonLws kvspicUtils websockets kvssdp kvsstun
+        ${AWSSDK_LINK_LIBRARIES})

cloudwatch-integ/Cloudwatch.cpp

@@ -0,0 +1,82 @@
+#include "Include.h"
+#include "Cloudwatch.h"
+
+namespace CppInteg {
+
+Cloudwatch::Cloudwatch(ClientConfiguration* pClientConfig)
+    : logs(pClientConfig), monitoring(pClientConfig), terminated(FALSE)
+{
+}
+
+STATUS Cloudwatch::init(PCHAR channelName, PCHAR region, BOOL isMaster, BOOL isStorage)
+{
+    ENTERS();
+    STATUS retStatus = STATUS_SUCCESS;
+    ClientConfiguration clientConfig;
+    CreateLogGroupRequest createLogGroupRequest;
+    Aws::CloudWatchLogs::Model::CreateLogStreamOutcome createLogStreamOutcome;
+    CreateLogStreamRequest createLogStreamRequest;
+
+    clientConfig.region = region;
+    auto& instance = getInstanceImpl(&clientConfig);
+
+    if (STATUS_FAILED(instance.logs.init(channelName, region, isMaster, isStorage))) {
+        DLOGW("Failed to create Cloudwatch logger");
+    } else {
+        globalCustomLogPrintFn = logger;
+    }
+
+    CHK_STATUS(instance.monitoring.init(channelName, region, isMaster, isStorage));
+
+CleanUp:
+
+    LEAVES();
+    return retStatus;
+}
+
+Cloudwatch& Cloudwatch::getInstance()
+{
+    return getInstanceImpl();
+}
+
+Cloudwatch& Cloudwatch::getInstanceImpl(ClientConfiguration* pClientConfig)
+{
+    static Cloudwatch instance{pClientConfig};
+    return instance;
+}
+
+VOID Cloudwatch::deinit()
+{
+    auto& instance = getInstance();
+    instance.logs.deinit();
+    instance.monitoring.deinit();
+    instance.terminated = TRUE;
+}
+
+VOID Cloudwatch::logger(UINT32 level, PCHAR tag, PCHAR fmt, ...)
+{
+    CHAR logFmtString[MAX_LOG_FORMAT_LENGTH + 1];
+    CHAR cwLogFmtString[MAX_LOG_FORMAT_LENGTH + 1];
+    UINT32 logLevel = GET_LOGGER_LOG_LEVEL();
+    UNUSED_PARAM(tag);
+
+    if (level >= logLevel) {
+        addLogMetadata(logFmtString, (UINT32) ARRAY_SIZE(logFmtString), fmt, level);
+
+        // Creating a copy to store the logFmtString for cloudwatch logging purpose
+        va_list valist, valist_cw;
+        va_start(valist_cw, fmt);
+        vsnprintf(cwLogFmtString, (SIZE_T) SIZEOF(cwLogFmtString), logFmtString, valist_cw);
+        va_end(valist_cw);
+        va_start(valist, fmt);
+        vprintf(logFmtString, valist);
+        va_end(valist);
+
+        auto& instance = getInstance();
+        if (!instance.terminated) {
+            instance.logs.push(cwLogFmtString);
+        }
+    }
+}
+
+} // namespace Canary

cloudwatch-integ/Cloudwatch.h

@@ -0,0 +1,31 @@
+#pragma once
+
+#include "Include.h"
+#include "CloudwatchLogs.h"
+#include "CloudwatchMonitoring.h"
+
+namespace CppInteg {
+
+class Cloudwatch {
+  public:
+    Cloudwatch() = delete;
+    Cloudwatch(Cloudwatch const&) = delete;
+    void operator=(Cloudwatch const&) = delete;
+
+    CloudwatchLogs logs;
+    CloudwatchMonitoring monitoring;
+
+    static Cloudwatch& getInstance();
+    static STATUS init(PCHAR channelName, PCHAR region, BOOL isMaster, BOOL isStorage);
+    static VOID deinit();
+    static VOID logger(UINT32, PCHAR, PCHAR, ...);
+
+  private:
+    static Cloudwatch& getInstanceImpl(ClientConfiguration* = nullptr);
+
+    Cloudwatch(ClientConfiguration*);
+    BOOL terminated;
+};
+typedef Cloudwatch* PCloudwatch;
+
+} // namespace Canary

cloudwatch-integ/CloudwatchLogs.cpp

@@ -0,0 +1,123 @@
+#include "Include.h"
+#include "CloudwatchLogs.h"
+
+namespace CppInteg {
+
+CloudwatchLogs::CloudwatchLogs(ClientConfiguration* pClientConfig) : client(*pClientConfig)
+{
+}
+
+STATUS CloudwatchLogs::init(PCHAR channelName, PCHAR region, BOOL isMaster, BOOL isStorage)
+{
+    STATUS retStatus = STATUS_SUCCESS;
+    CreateLogGroupRequest createLogGroupRequest;
+    Aws::CloudWatchLogs::Model::CreateLogStreamOutcome createLogStreamOutcome;
+    CreateLogStreamRequest createLogStreamRequest;
+    std::stringstream defaultLogStreamName;
+    if(isStorage) {
+        defaultLogStreamName << channelName << '-' << "StorageMaster" << '-'
+                             << GETTIME() / HUNDREDS_OF_NANOS_IN_A_MILLISECOND;
+
+    } else {
+        defaultLogStreamName << channelName << '-' << (isMaster ? "master" : "viewer") << '-'
+                             << GETTIME() / HUNDREDS_OF_NANOS_IN_A_MILLISECOND;
+
+    }
+
+    this->logStreamName = defaultLogStreamName.str();
+    this->logGroupName = LOG_GROUP_NAME;
+
+    DLOGI("Log stream name: %s", this->logStreamName.c_str());
+
+    createLogGroupRequest.SetLogGroupName(this->logGroupName);
+    // ignore error since if this operation fails, CreateLogStream should fail as well.
+    // There might be some errors that can lead to successfull CreateLogStream, e.g. log group already exists.
+    this->client.CreateLogGroup(createLogGroupRequest);
+
+    createLogStreamRequest.SetLogGroupName(this->logGroupName);
+    createLogStreamRequest.SetLogStreamName(this->logStreamName);
+    createLogStreamOutcome = this->client.CreateLogStream(createLogStreamRequest);
+
+    CHK_ERR(createLogStreamOutcome.IsSuccess(), STATUS_INVALID_OPERATION, "Failed to create \"%s\" log stream: %s",
+            this->logStreamName.c_str(), createLogStreamOutcome.GetError().GetMessage().c_str());
+
+CleanUp:
+
+    return retStatus;
+}
+
+VOID CloudwatchLogs::deinit()
+{
+    this->flush(TRUE);
+}
+
+VOID CloudwatchLogs::push(string log)
+{
+    std::lock_guard<std::recursive_mutex> lock(this->sync.mutex);
+    Aws::String awsCwString(log.c_str(), log.size());
+    auto logEvent =
+        Aws::CloudWatchLogs::Model::InputLogEvent().WithMessage(awsCwString).WithTimestamp(GETTIME() / HUNDREDS_OF_NANOS_IN_A_MILLISECOND);
+    this->logs.push_back(logEvent);
+    if (this->logs.size() >= MAX_CLOUDWATCH_LOG_COUNT) {
+        this->flush();
+    }
+}
+
+VOID CloudwatchLogs::flush(BOOL sync)
+{
+    std::unique_lock<std::recursive_mutex> lock(this->sync.mutex);
+    if (this->logs.size() == 0) {
+        return;
+    }
+    auto pendingLogs = this->logs;
+    this->logs.clear();
+
+    // wait until previous logs have been flushed entirely
+    auto waitUntilFlushed = [this] { return !this->sync.pending.load(); };
+    this->sync.await.wait(lock, waitUntilFlushed);
+
+    auto request = Aws::CloudWatchLogs::Model::PutLogEventsRequest()
+                       .WithLogGroupName(this->logGroupName)
+                       .WithLogStreamName(this->logStreamName)
+                       .WithLogEvents(pendingLogs);
+
+    if (this->token != "") {
+        request.SetSequenceToken(this->token);
+    }
+
+    if (!sync) {
+        auto asyncHandler = [this](const Aws::CloudWatchLogs::CloudWatchLogsClient* cwClientLog,
+                                   const Aws::CloudWatchLogs::Model::PutLogEventsRequest& request,
+                                   const Aws::CloudWatchLogs::Model::PutLogEventsOutcome& outcome,
+                                   const std::shared_ptr<const Aws::Client::AsyncCallerContext>& context) {
+            UNUSED_PARAM(cwClientLog);
+            UNUSED_PARAM(request);
+            UNUSED_PARAM(context);
+
+            if (!outcome.IsSuccess()) {
+                // Need to use printf so that we don't get into an infinite loop where we keep flushing
+                printf("Failed to push logs: %s\n", outcome.GetError().GetMessage().c_str());
+            } else {
+                printf("Successfully pushed logs to cloudwatch\n");
+                this->token = outcome.GetResult().GetNextSequenceToken();
+            }
+
+            this->sync.pending = FALSE;
+            this->sync.await.notify_one();
+        };
+
+        this->sync.pending = TRUE;
+        this->client.PutLogEventsAsync(request, asyncHandler);
+    } else {
+        auto outcome = this->client.PutLogEvents(request);
+        if (!outcome.IsSuccess()) {
+            // Need to use printf so that we don't get into an infinite loop where we keep flushing
+            printf("Failed to push logs: %s\n", outcome.GetError().GetMessage().c_str());
+        } else {
+            DLOGS("Successfully pushed logs to cloudwatch");
+            this->token = outcome.GetResult().GetNextSequenceToken();
+        }
+    }
+}
+
+} // namespace Canary