Merge pull request #2528 from newrelic/sophocles

serverless updates, including support for v1 payload formats

.rubocop.yml

@@ -1249,6 +1249,7 @@ Style/MethodCallWithArgsParentheses:
     - add_dependency
     - add_development_dependency
     - catch
+    - debug
     - expect
     - fail
     - gem

lib/new_relic/agent.rb

@@ -620,16 +620,19 @@ def tl_is_execution_traced?
     def add_custom_attributes(params) # THREAD_LOCAL_ACCESS
       record_api_supportability_metric(:add_custom_attributes)
 
-      if params.is_a?(Hash)
-        Transaction.tl_current&.add_custom_attributes(params)
-
-        segment = ::NewRelic::Agent::Tracer.current_segment
-        if segment
-          add_new_segment_attributes(params, segment)
-        end
-      else
+      unless params.is_a?(Hash)
         ::NewRelic::Agent.logger.warn("Bad argument passed to #add_custom_attributes. Expected Hash but got #{params.class}")
+        return
       end
+
+      if NewRelic::Agent.agent&.serverless?
+        ::NewRelic::Agent.logger.warn('Custom attributes are not supported in serverless mode')
+        return
+      end
+
+      Transaction.tl_current&.add_custom_attributes(params)
+      segment = ::NewRelic::Agent::Tracer.current_segment
+      add_new_segment_attributes(params, segment) if segment
     end
 
     def add_new_segment_attributes(params, segment)

lib/new_relic/agent/new_relic_service.rb

@@ -143,6 +143,9 @@ def build_metric_data_array(stats_hash)
       end
 
       def metric_data(stats_hash)
+        # let the serverless handler handle serialization
+        return NewRelic::Agent.agent.serverless_handler.metric_data(stats_hash) if NewRelic::Agent.agent.serverless?
+
         timeslice_start = stats_hash.started_at
         timeslice_end = stats_hash.harvested_at || Process.clock_gettime(Process::CLOCK_REALTIME)
         metric_data_array = build_metric_data_array(stats_hash)
@@ -154,6 +157,9 @@ def metric_data(stats_hash)
       end
 
       def error_data(unsent_errors)
+        # let the serverless handler handle serialization
+        return NewRelic::Agent.agent.serverless_handler.error_data(unsent_errors) if NewRelic::Agent.agent.serverless?
+
         invoke_remote(:error_data, [@agent_id, unsent_errors],
           :item_count => unsent_errors.size)
       end

lib/new_relic/agent/serverless_handler.rb

@@ -8,34 +8,37 @@
 module NewRelic
   module Agent
     class ServerlessHandler
-      COLD_START_ATTRIBUTE = 'aws.lambda.coldStart'
-      COLD_START_DESTINATIONS = NewRelic::Agent::AttributeFilter::DST_TRANSACTION_TRACER |
+      ATTRIBUTE_ARN = 'aws.lambda.arn'
+      ATTRIBUTE_COLD_START = 'aws.lambda.coldStart'
+      ATTRIBUTE_REQUEST_ID = 'aws.requestId'
+      AGENT_ATTRIBUTE_DESTINATIONS = NewRelic::Agent::AttributeFilter::DST_TRANSACTION_TRACER |
         NewRelic::Agent::AttributeFilter::DST_TRANSACTION_EVENTS
       EXECUTION_ENVIRONMENT = "AWS_Lambda_ruby#{RUBY_VERSION.rpartition('.').first}".freeze
       LAMBDA_MARKER = 'NR_LAMBDA_MONITORING'
       LAMBDA_ENVIRONMENT_VARIABLE = 'AWS_LAMBDA_FUNCTION_NAME'
-      METADATA_VERSION = 2 # internal to New Relic's cross-agent specs
-      METHOD_BLOCKLIST = %i[connect preconnect shutdown profile_data get_agent_commands agent_command_results].freeze
+      METHOD_BLOCKLIST = %i[agent_command_results connect get_agent_commands preconnect profile_data
+        shutdown].freeze
       NAMED_PIPE = '/tmp/newrelic-telemetry'
       SUPPORTABILITY_METRIC = 'Supportability/AWSLambda/HandlerInvocation'
       FUNCTION_NAME = 'lambda_function'
-      VERSION = 2 # internal to New Relic's cross-agent specs
+      PAYLOAD_VERSION = ENV.fetch('NEW_RELIC_SERVERLESS_PAYLOAD_VERSION', 2)
 
       def self.env_var_set?
         ENV.key?(LAMBDA_ENVIRONMENT_VARIABLE)
       end
 
       def initialize
+        @context = nil
         @payloads = {}
       end
 
       def invoke_lambda_function_with_new_relic(event:, context:, method_name:, namespace: nil)
         NewRelic::Agent.increment_metric(SUPPORTABILITY_METRIC)
 
-        parse_context(context)
+        @context = context
 
         NewRelic::Agent::Tracer.in_transaction(category: :other, name: function_name) do
-          notice_cold_start
+          add_agent_attributes
 
           NewRelic::LanguageSupport.constantize(namespace).send(method_name, event: event, context: context)
         end
@@ -51,6 +54,36 @@ def store_payload(method, payload)
         @payloads[method] = payload
       end
 
+      def metric_data(stats_hash)
+        payload = [nil,
+          stats_hash.started_at,
+          (stats_hash.harvested_at || Process.clock_gettime(Process::CLOCK_REALTIME)),
+          []]
+        stats_hash.each do |metric_spec, stats|
+          next if stats.is_reset?
+
+          hash = {name: metric_spec.name}
+          hash[:scope] = metric_spec.scope unless metric_spec.scope.empty?
+
+          payload.last.push([hash, [
+            stats.call_count,
+            stats.total_call_time,
+            stats.total_exclusive_time,
+            stats.min_call_time,
+            stats.max_call_time,
+            stats.sum_of_squares
+          ]])
+        end
+
+        return if payload.last.empty?
+
+        store_payload(:metric_data, payload)
+      end
+
+      def error_data(errors)
+        store_payload(:error_data, [nil, errors.map(&:to_collector_array)])
+      end
+
       private
 
       def harvest!
@@ -60,39 +93,43 @@ def harvest!
       end
 
       def metadata
-        {arn: @function_arn,
+        {arn: @context.invoked_function_arn,
          protocol_version: NewRelic::Agent::NewRelicService::PROTOCOL_VERSION,
-         function_version: @function_version,
+         function_version: @context.function_version,
          execution_environment: EXECUTION_ENVIRONMENT,
-         agent_version: NewRelic::VERSION::STRING,
-         metadata_version: METADATA_VERSION,
-         agent_language: LANGUAGE}.reject { |_k, v| v.nil? }
-      end
-
-      def parse_context(context)
-        @function_arn = nil
-        @function_version = nil
-        return unless context
-        return unless context.respond_to?(:function_arn) && context.respond_to?(:function_version)
-
-        @function_arn = context.function_arn
-        @function_version = context.function_version
+         agent_version: NewRelic::VERSION::STRING}
       end
 
       def function_name
         ENV.fetch(LAMBDA_ENVIRONMENT_VARIABLE, FUNCTION_NAME)
       end
 
       def write_output
-        json = NewRelic::Agent.agent.service.marshaller.dump(@payloads)
-        gzipped = NewRelic::Agent::NewRelicService::Encoders::Compressed::Gzip.encode(json)
-        base64_encoded = NewRelic::Base64.encode64(gzipped)
-        array = [VERSION, LAMBDA_MARKER, metadata, base64_encoded]
-        string = ::JSON.dump(array)
+        string = PAYLOAD_VERSION == 1 ? payload_v1 : payload_v2
 
         return puts string unless use_named_pipe?
 
-        File.open(NAMED_PIPE, 'w') { |f| f.puts string }
+        File.write(NAMED_PIPE, string)
+
+        NewRelic::Agent.logger.debug "Wrote serverless payload to #{NAMED_PIPE}\n" \
+          "BEGIN PAYLOAD>>>\n#{string}\n<<<END PAYLOAD"
+      end
+
+      def payload_v1
+        payload_hash = {'metadata' => metadata, 'data' => @payloads}
+        json = NewRelic::Agent.agent.service.marshaller.dump(payload_hash)
+        gzipped = NewRelic::Agent::NewRelicService::Encoders::Compressed::Gzip.encode(json)
+        base64_encoded = NewRelic::Base64.strict_encode64(gzipped)
+        array = [PAYLOAD_VERSION, LAMBDA_MARKER, base64_encoded]
+        ::JSON.dump(array)
+      end
+
+      def payload_v2
+        json = NewRelic::Agent.agent.service.marshaller.dump(@payloads)
+        gzipped = NewRelic::Agent::NewRelicService::Encoders::Compressed::Gzip.encode(json)
+        base64_encoded = NewRelic::Base64.strict_encode64(gzipped)
+        array = [PAYLOAD_VERSION, LAMBDA_MARKER, metadata, base64_encoded]
+        ::JSON.dump(array)
       end
 
       def use_named_pipe?
@@ -101,12 +138,16 @@ def use_named_pipe?
         @use_named_pipe = File.exist?(NAMED_PIPE) && File.writable?(NAMED_PIPE)
       end
 
-      def notice_cold_start
-        return unless cold? && NewRelic::Agent::Tracer.current_transaction
+      def add_agent_attributes
+        return unless NewRelic::Agent::Tracer.current_transaction
+
+        add_agent_attribute(ATTRIBUTE_COLD_START, true) if cold?
+        add_agent_attribute(ATTRIBUTE_ARN, @context.invoked_function_arn)
+        add_agent_attribute(ATTRIBUTE_REQUEST_ID, @context.aws_request_id)
+      end
 
-        NewRelic::Agent::Tracer.current_transaction.add_agent_attribute(COLD_START_ATTRIBUTE,
-          true,
-          COLD_START_DESTINATIONS)
+      def add_agent_attribute(attribute, value)
+        NewRelic::Agent::Tracer.current_transaction.add_agent_attribute(attribute, value, AGENT_ATTRIBUTE_DESTINATIONS)
       end
 
       def cold?
@@ -117,6 +158,7 @@ def cold?
       end
 
       def reset!
+        @context = nil
         @payloads.replace({})
       end
     end