WIP

components/services/Gemfile

@@ -16,6 +16,9 @@ gem "sequel"
 
 group :development do
   gem "rake"
+  gem "rubocop"
+  gem "rubocop-rspec"
+  gem "rubocop-performance"
 end
 
 group :test do

components/services/Gemfile.lock

@@ -9,6 +9,7 @@ GIT
 GEM
   remote: https://rubygems.org/
   specs:
+    ast (2.4.2)
     aws-eventstream (1.3.0)
     aws-partitions (1.976.0)
     aws-sdk-core (3.206.0)
@@ -34,14 +35,23 @@ GEM
     diff-lcs (1.5.1)
     docile (1.4.1)
     jmespath (1.6.2)
+    json (2.7.2)
+    language_server-protocol (3.17.0.3)
     method_source (1.1.0)
     ostruct (0.6.0)
     ox (2.14.18)
+    parallel (1.26.3)
+    parser (3.3.5.0)
+      ast (~> 2.4.1)
+      racc
     pg (1.5.8)
     pry (0.14.2)
       coderay (~> 1.1)
       method_source (~> 1.0)
+    racc (1.8.1)
+    rainbow (3.1.1)
     rake (13.2.1)
+    regexp_parser (2.9.2)
     rexml (3.3.7)
     rspec (3.13.0)
       rspec-core (~> 3.13.0)
@@ -56,6 +66,24 @@ GEM
       diff-lcs (>= 1.2.0, < 2.0)
       rspec-support (~> 3.13.0)
     rspec-support (3.13.1)
+    rubocop (1.66.1)
+      json (~> 2.3)
+      language_server-protocol (>= 3.17.0)
+      parallel (~> 1.10)
+      parser (>= 3.3.0.2)
+      rainbow (>= 2.2.2, < 4.0)
+      regexp_parser (>= 2.4, < 3.0)
+      rubocop-ast (>= 1.32.2, < 2.0)
+      ruby-progressbar (~> 1.7)
+      unicode-display_width (>= 2.4.0, < 3.0)
+    rubocop-ast (1.32.3)
+      parser (>= 3.3.1.0)
+    rubocop-performance (1.22.1)
+      rubocop (>= 1.48.1, < 2.0)
+      rubocop-ast (>= 1.31.1, < 2.0)
+    rubocop-rspec (3.0.5)
+      rubocop (~> 1.61)
+    ruby-progressbar (1.13.0)
     sentry-ruby (5.19.0)
       bigdecimal
       concurrent-ruby (~> 1.0, >= 1.0.2)
@@ -70,6 +98,7 @@ GEM
       simplecov (~> 0.19)
     simplecov-html (0.13.1)
     simplecov_json_formatter (0.1.4)
+    unicode-display_width (2.6.0)
 
 PLATFORMS
   ruby
@@ -85,6 +114,9 @@ DEPENDENCIES
   pry
   rake
   rspec
+  rubocop
+  rubocop-performance
+  rubocop-rspec
   sentry-ruby
   sequel
   simplecov

components/services/app.rb

@@ -10,7 +10,6 @@ def self.process(event:, context:)
       logger = Logger.new($stdout)
       logger.info("## Processing Event")
       logger.info(event)
-      logger.info(context)
 
       new(event:, context:).process
     rescue Exception => e
@@ -31,6 +30,8 @@ def process
         HandleSQSMessageEvent.call(event:)
       when :service_action
         event.service_action.call(**event.parameters)
+      when :cloudwatch_log_event
+        HandleLogEvents.call(event:)
       end
     end
 

components/services/app/parsers/cloudwatch_log_event_parser.rb

@@ -0,0 +1,55 @@
+
+require "zlib"
+require "base64"
+require "stringio"
+
+class CloudWatchLogEventParser
+  LogEvent = Struct.new(
+    :timestamp,
+    :message,
+  )
+
+  Event = Struct.new(
+    :event_type,
+    :log_group,
+    :log_events,
+    keyword_init: true
+  )
+
+  attr_reader :event
+
+  def initialize(event)
+    @event = event
+  end
+
+  def parse_event
+    Event.new(
+      event_type: :cloudwatch_log_event,
+      log_group:,
+      log_events:
+    )
+  end
+
+  private
+
+  def raw_data
+    event.dig("awslogs", "data")
+  end
+
+  def data_message
+    @data_message ||= JSON.parse(Zlib::GzipReader.new(StringIO.new(Base64.decode64(raw_data))).read)
+  end
+
+  def log_group
+    data_message.fetch("logGroup")
+  end
+
+  def log_events
+    data_message.fetch("logEvents").map do |log_event_data|
+      LogEvent.new(
+        timestamp: Time.at(log_event_data.fetch("timestamp") / 1000),
+        message: log_event_data.fetch("message")
+      )
+    end
+  end
+end

components/services/app/parsers/event_parser.rb

@@ -12,6 +12,8 @@ def parse_event
                SQSMessageEventParser
     elsif service_action?
                ServiceActionParser
+    elsif cloudwatch_log_event?
+               CloudWatchLogEventParser
     end
 
     parser.new(event).parse_event
@@ -30,4 +32,8 @@ def ecs_event?
   def service_action?
     event.key?("serviceAction")
   end
+
+  def cloudwatch_log_event?
+    event.keys.size == 1 && event.key?("awslogs")
+  end
 end

components/services/app/parsers/opensips_log_event_parser.rb

@@ -0,0 +1,17 @@
+class OpenSIPSLogEventParser
+  attr_reader :event
+
+  Event = Struct.new(:level, :message, keyword_init: true)
+
+  def initialize(event)
+    @event = event
+  end
+
+  def parse_event
+    event.log_events.map do |log_event|
+      log_data = JSON.parse(log_event.message)
+
+      Event.new(level: log_data.fetch("level"), message: log_data.fetch("message"))
+    end
+  end
+end

components/services/app/workflows/handle_log_events.rb

@@ -0,0 +1,23 @@
+require "zlib"
+require "base64"
+require "stringio"
+
+class HandleLogEvents < ApplicationWorkflow
+  attr_reader :event
+
+  def initialize(event:)
+    @event = event
+  end
+
+  def call
+    if opensips_log_groups.include?(event.log_group)
+      HandleOpenSIPSLogEvent.call(event:)
+    end
+  end
+
+  private
+
+  def opensips_log_groups
+    [ ENV.fetch("PUBLIC_GATEWAY_LOG_GROUP"), ENV.fetch("CLIENT_GATEWAY_LOG_GROUP") ]
+  end
+end

components/services/app/workflows/handle_opensips_log_event.rb

@@ -0,0 +1,47 @@
+class HandleOpenSIPSLogEvent < ApplicationWorkflow
+  LOAD_BALANCER_RESPONSE_ERROR_PATTERN = %r{\A(\d{3})-lb-response-error-(((25[0-5]|(2[0-4]|1\d|[1-9]|)\d)\.?\b){4})\z}
+  TIMEOUT_ERROR_CODE = "408"
+
+  attr_reader :event, :error_tracking_client
+
+  LoadBalancerError = Struct.new(:target_ip, :code, keyword_init: true)
+
+  def initialize(event:, error_tracking_client: Sentry)
+    @event = event
+    @error_tracking_client = error_tracking_client
+  end
+
+  def call
+    if load_balancer_response_errors.any?
+      error_messages = load_balancer_response_errors.map do |error|
+        "Error detected on load balancer: #{error.target_ip} - #{error.code}"
+      end
+      error_tracking_client.capture_message(error_messages.join("\n"))
+    else
+      error_messages = opensips_logs.each do |log|
+        log.message
+      end
+
+      error_tracking_client.capture_message(error_messages.join("\n"))
+    end
+  end
+
+  def opensips_logs
+    @opensips_logs ||= OpenSIPSLogEventParser.new(event).parse_event
+  end
+
+  def load_balancer_response_errors
+    errors = opensips_logs.select do |log|
+      log.message.match?(LOAD_BALANCER_RESPONSE_ERROR_PATTERN)
+    end
+
+    errors.map do |log|
+      code, ip, = LOAD_BALANCER_RESPONSE_ERROR_PATTERN.match(log.message).captures
+      LoadBalancerError.new(target_ip: IPAddr.new(ip), code:)
+    end
+  end
+
+  def find_errors_by_code(code)
+    load_balancer_response_errors.select { |error| error.code == code }
+  end
+end

components/services/spec/fixtures/files/cloudwatch_log_event.json

@@ -0,0 +1,5 @@
+{
+  "awslogs": {
+    "data": "H4sIAAAAAAAA/3WQy2rDMBBFf8XM2g4aWYoeO0PdbNpNnV1dipMMRuCHkJSEEPLvxXG77HLuPRwuc4eRYux62t88gYWXal99v9dNU+1qyGG+ThTAQskFV2ZbbiVTkMMw97swnz1Y8OfD4I5F3yW6drcipq53U78yTQrUjWAhUUxrGs+HeAzOJzdPr25IFCLYz/8sX09NfaEpLdgd3GkZoyWWaBBRCKkkMmUUU9wIUyqGUkpWasMEcimYlqhRK8MMcsghuZFi6kYPFhXfGoMcNVM6/3sDWLi3T6wFm7XQkM84z5iyXFi+bSHPWvDutLSaL8dAFxpWuHqrP/Yr8qtbc8F0MRyKQNHPU6SCQphDgYpvUG/YRrbwgMfX4wcsVdvmjAEAAA=="
+  }
+}

components/services/spec/requests/cloudwatch_log_events_spec.rb

@@ -0,0 +1,33 @@
+require_relative "../spec_helper"
+
+RSpec.describe "Handle CloudWatch Log Events" do
+  it "handles public gateway alerts" do
+    stub_env("PUBLIC_GATEWAY_LOG_GROUP" => "public-gateway")
+    payload = build_cloudwatch_log_event_payload(
+      log_group: "public-gateway",
+      log_events: [
+        build_cloudwatch_log_event(
+          message: build_opensips_message(message: "408-lb-response-error-10.10.1.180")
+        )
+      ]
+    )
+
+    invoke_lambda(payload:)
+  end
+
+  def build_opensips_message(data = {})
+    data = {
+      time: "Sep 22 07:24:26",
+      pid: 82,
+      level: "CRITICAL",
+      message: "error"
+    }.merge(data)
+
+    {
+      "time" => data.fetch(:time),
+      "pid" => data.fetch(:pid),
+      "level" => data.fetch(:level),
+      "message" => data.fetch(:message)
+    }.to_json
+  end
+end

components/services/spec/spec_helper.rb

@@ -24,6 +24,8 @@
   SimpleCov.formatter = SimpleCov::Formatter::CoberturaFormatter
 end
 
+ENV["PUBLIC_GATEWAY_LOG_GROUP"] = "public-gateway"
+ENV["CLIENT_GATEWAY_LOG_GROUP"] = "client-gateway"
 ENV["SWITCH_GROUP"] = "service:somleng-switch"
 ENV["MEDIA_PROXY_GROUP"] = "service:media-proxy"
 ENV["CLIENT_GATEWAY_GROUP"] = "service:client-gateway"

components/services/spec/support/event_helpers.rb

@@ -1,6 +1,48 @@
 require "json"
 
 module EventHelpers
+  def build_cloudwatch_log_event(data = {})
+    data = {
+      id: SecureRandom.random_number(10**56).to_s.rjust(56, "0"),
+      timestamp: Time.now.to_i * 1000,
+      message: "log-message"
+    }.merge(data)
+
+    {
+      "id" => data.fetch(:id),
+      "timestamp" => data.fetch(:timestamp),
+      "message" => data.fetch(:message)
+    }
+  end
+
+  def build_cloudwatch_log_event_payload(data = {})
+    data = {
+      log_group: "testing",
+      log_events: [
+        build_cloudwatch_log_event
+      ]
+    }.merge(data)
+
+    payload = JSON.parse(file_fixture("cloudwatch_log_event.json").read)
+    data_message = JSON.parse(Zlib::GzipReader.new(StringIO.new(Base64.decode64(payload.dig("awslogs", "data")))).read)
+
+    overrides = {
+      "logGroup" => data.fetch(:log_group),
+      "logEvents" => data.fetch(:log_events)
+    }
+
+    compressed_data_message = StringIO.new
+    gz = Zlib::GzipWriter.new(compressed_data_message)
+    gz.write(data_message.merge(overrides).to_json)
+    gz.close
+
+    {
+      "awslogs" => {
+        "data" => Base64.encode64(compressed_data_message.string)
+      }
+    }
+  end
+
   def build_sqs_message_event_payload(data = {})
     data = {
       event_source_arn: "arn:aws:sqs:us-east-2:123456789012:somleng-switch-permissions",

infrastructure/modules/services/lambda.tf

@@ -14,6 +14,8 @@ resource "aws_lambda_function" "this" {
 
   environment {
     variables = {
+      PUBLIC_GATEWAY_LOG_GROUP                    = var.public_gateway_group
+      CLIENT_GATEWAY_LOG_GROUP                    = var.client_gateway_group
       SWITCH_GROUP                                = "service:${var.switch_group}"
       MEDIA_PROXY_GROUP                           = "service:${var.media_proxy_group}"
       CLIENT_GATEWAY_GROUP                        = "service:${var.client_gateway_group}"

infrastructure/modules/services/variables.tf

@@ -4,6 +4,7 @@ variable "app_image" {}
 variable "vpc" {}
 variable "switch_group" {}
 variable "media_proxy_group" {}
+variable "public_gateway_group" {}
 variable "client_gateway_group" {}
 variable "db_password_parameter" {}
 variable "freeswitch_event_socket_password_parameter" {}

infrastructure/staging/services.tf

@@ -5,6 +5,7 @@ module "services" {
   app_environment        = var.app_environment
   switch_group           = var.switch_identifier
   media_proxy_group      = var.media_proxy_identifier
+  public_gateway_group   = var.public_gateway_identifier
   client_gateway_group   = var.client_gateway_identifier
   public_gateway_db_name = var.public_gateway_db_name
   client_gateway_db_name = var.client_gateway_db_name