Changing add_event to start sending missing fields

azurelinuxagent/common/event.py

@@ -20,6 +20,7 @@
 import json
 import os
 import sys
+import threading
 import time
 import traceback
 from datetime import datetime
@@ -31,14 +32,19 @@
 from azurelinuxagent.common.datacontract import get_properties
 from azurelinuxagent.common.telemetryevent import TelemetryEventParam, TelemetryEvent
 from azurelinuxagent.common.utils import fileutil, textutil
-from azurelinuxagent.common.version import CURRENT_VERSION
+from azurelinuxagent.common.version import CURRENT_VERSION, CURRENT_AGENT
 
 _EVENT_MSG = "Event: name={0}, op={1}, message={2}, duration={3}"
 TELEMETRY_EVENT_PROVIDER_ID = "69B669B9-4AF8-4C50-BDC4-6006FA76E975"
 
 # Store the last retrieved container id as an environment variable to be shared between threads for telemetry purposes
 CONTAINER_ID_ENV_VARIABLE = "AZURE_GUEST_AGENT_CONTAINER_ID"
 
+
+def get_container_id_from_env():
+    return os.environ.get(CONTAINER_ID_ENV_VARIABLE, "UNINITIALIZED")
+
+
 class WALAEventOperation:
     ActivateResourceDisk = "ActivateResourceDisk"
     AgentBlacklisted = "AgentBlacklisted"
@@ -244,30 +250,19 @@ def is_period_elapsed(self, delta, h):
         return h not in self.periodic_events or \
             (self.periodic_events[h] + delta) <= datetime.now()
 
-    def add_periodic(self,
-                     delta, name, op=WALAEventOperation.Unknown, is_success=True, duration=0,
-                     version=CURRENT_VERSION, message="", evt_type="",
-                     is_internal=False, log_event=True, force=False):
-
+    def add_periodic(self, delta, name, op=WALAEventOperation.Unknown, is_success=True, duration=0,
+                     version=str(CURRENT_VERSION), message="", evt_type="", is_internal=False, log_event=True,
+                     force=False):
         h = hash(name + op + ustr(is_success) + message)
 
         if force or self.is_period_elapsed(delta, h):
-            self.add_event(name,
-                           op=op, is_success=is_success, duration=duration,
+            self.add_event(name, op=op, is_success=is_success, duration=duration,
                            version=version, message=message, evt_type=evt_type,
                            is_internal=is_internal, log_event=log_event)
             self.periodic_events[h] = datetime.now()
 
-    def add_event(self,
-                  name,
-                  op=WALAEventOperation.Unknown,
-                  is_success=True,
-                  duration=0,
-                  version=CURRENT_VERSION,
-                  message="",
-                  evt_type="",
-                  is_internal=False,
-                  log_event=True):
+    def add_event(self, name, op=WALAEventOperation.Unknown, is_success=True, duration=0, version=str(CURRENT_VERSION),
+                  message="", evt_type="", is_internal=False, log_event=True):
 
         if (not is_success) and log_event:
             _log_event(name, op, message, duration, is_success=is_success)
@@ -280,14 +275,12 @@ def _add_event(self, duration, evt_type, is_internal, is_success, message, name,
         event.parameters.append(TelemetryEventParam('Version', str(version)))
         event.parameters.append(TelemetryEventParam('IsInternal', is_internal))
         event.parameters.append(TelemetryEventParam('Operation', op))
-        event.parameters.append(TelemetryEventParam('OperationSuccess',
-                                                    is_success))
+        event.parameters.append(TelemetryEventParam('OperationSuccess', is_success))
         event.parameters.append(TelemetryEventParam('Message', message))
         event.parameters.append(TelemetryEventParam('Duration', duration))
         event.parameters.append(TelemetryEventParam('ExtensionType', evt_type))
-        event.parameters.append(TelemetryEventParam('ContainerId',
-                                                    os.environ.get(CONTAINER_ID_ENV_VARIABLE, "UNINITIALIZED")))
 
+        self._add_default_parameters_to_event(event)
         data = get_properties(event)
         try:
             self.save_event(json.dumps(data))
@@ -314,6 +307,7 @@ def add_log_event(self, level, message):
         event.parameters.append(TelemetryEventParam('Context2', ''))
         event.parameters.append(TelemetryEventParam('Context3', ''))
 
+        self._add_default_parameters_to_event(event)
         data = get_properties(event)
         try:
             self.save_event(json.dumps(data))
@@ -341,12 +335,30 @@ def add_metric(self, category, counter, instance, value, log_event=False):
         event.parameters.append(TelemetryEventParam('Instance', instance))
         event.parameters.append(TelemetryEventParam('Value', value))
 
+        self._add_default_parameters_to_event(event)
         data = get_properties(event)
         try:
             self.save_event(json.dumps(data))
         except EventError as e:
             logger.error("{0}", e)
 
+    @staticmethod
+    def _add_default_parameters_to_event(event):
+        task_name = threading.current_thread().getName()
+
+        # We set thread names for the threads spawned by ExtHandler. We get the name of the thread,
+        # and if it is "MainThread" (the default name from Python. ref threading._MainThread),
+        # then we default to "ExtHandler".
+        task_name = task_name if task_name != "MainThread" else "ExtHandler"
+
+        event.parameters.append(TelemetryEventParam("GAVersion", CURRENT_AGENT))
+        event.parameters.append(TelemetryEventParam('ContainerId', get_container_id_from_env()))
+        event.parameters.append(TelemetryEventParam('OpcodeName', datetime.utcnow().__str__()))
+        event.parameters.append(TelemetryEventParam('EventTid', threading.current_thread().ident))
+        event.parameters.append(TelemetryEventParam('EventPid', os.getpid()))
+        event.parameters.append(TelemetryEventParam("TaskName", task_name))
+        event.parameters.append(TelemetryEventParam("KeywordName", ''))  # Unused field for now.
+
 
 __event_logger__ = EventLogger()
 
@@ -364,7 +376,7 @@ def elapsed_milliseconds(utc_start):
 def report_event(op, is_success=True, message='', log_event=True):
     from azurelinuxagent.common.version import AGENT_NAME, CURRENT_VERSION
     add_event(AGENT_NAME,
-              version=CURRENT_VERSION,
+              version=str(CURRENT_VERSION),
               is_success=is_success,
               message=message,
               op=op,
@@ -374,10 +386,10 @@ def report_event(op, is_success=True, message='', log_event=True):
 def report_periodic(delta, op, is_success=True, message=''):
     from azurelinuxagent.common.version import AGENT_NAME, CURRENT_VERSION
     add_periodic(delta, AGENT_NAME,
-              version=CURRENT_VERSION,
-              is_success=is_success,
-              message=message,
-              op=op)
+                 version=str(CURRENT_VERSION),
+                 is_success=is_success,
+                 message=message,
+                 op=op)
 
 
 def report_metric(category, counter, instance, value, log_event=False, reporter=__event_logger__):
@@ -399,21 +411,17 @@ def report_metric(category, counter, instance, value, log_event=False, reporter=
     reporter.add_metric(category, counter, instance, value, log_event)
 
 
-def add_event(name, op=WALAEventOperation.Unknown, is_success=True, duration=0,
-              version=CURRENT_VERSION,
-              message="", evt_type="", is_internal=False, log_event=True,
-              reporter=__event_logger__):
+def add_event(name, op=WALAEventOperation.Unknown, is_success=True, duration=0, version=str(CURRENT_VERSION), message="",
+              evt_type="", is_internal=False, log_event=True, reporter=__event_logger__):
     if reporter.event_dir is None:
         logger.warn("Cannot add event -- Event reporter is not initialized.")
         _log_event(name, op, message, duration, is_success=is_success)
         return
 
     if should_emit_event(name, version, op, is_success):
         mark_event_status(name, version, op, is_success)
-        reporter.add_event(
-            name, op=op, is_success=is_success, duration=duration,
-            version=str(version), message=message, evt_type=evt_type,
-            is_internal=is_internal, log_event=log_event)
+        reporter.add_event(name, op=op, is_success=is_success, duration=duration, version=str(version), message=message,
+                           evt_type=evt_type, is_internal=is_internal, log_event=log_event)
 
 
 def add_log_event(level, message, reporter=__event_logger__):
@@ -423,20 +431,16 @@ def add_log_event(level, message, reporter=__event_logger__):
     reporter.add_log_event(level, message)
 
 
-def add_periodic(
-    delta, name, op=WALAEventOperation.Unknown, is_success=True, duration=0,
-    version=CURRENT_VERSION,
-    message="", evt_type="", is_internal=False, log_event=True, force=False,
-    reporter=__event_logger__):
+def add_periodic(delta, name, op=WALAEventOperation.Unknown, is_success=True, duration=0,
+                 version=str(CURRENT_VERSION), message="", evt_type="", is_internal=False, log_event=True, force=False,
+                 reporter=__event_logger__):
     if reporter.event_dir is None:
         logger.warn("Cannot add periodic event -- Event reporter is not initialized.")
         _log_event(name, op, message, duration, is_success=is_success)
         return
 
-    reporter.add_periodic(
-        delta, name, op=op, is_success=is_success, duration=duration,
-        version=str(version), message=message, evt_type=evt_type,
-        is_internal=is_internal, log_event=log_event, force=force)
+    reporter.add_periodic(delta, name, op=op, is_success=is_success, duration=duration, version=str(version),
+                          message=message, evt_type=evt_type, is_internal=is_internal, log_event=log_event, force=force)
 
 
 def mark_event_status(name, version, op, status):

azurelinuxagent/ga/env.py

@@ -88,6 +88,7 @@ def is_alive(self):
     def start(self):
         self.server_thread = threading.Thread(target=self.monitor)
         self.server_thread.setDaemon(True)
+        self.server_thread.setName("EnvHandler")
         self.server_thread.start()
 
     def monitor(self):

azurelinuxagent/ga/monitor.py

@@ -162,6 +162,7 @@ def is_alive(self):
     def start(self):
         self.event_thread = threading.Thread(target=self.daemon)
         self.event_thread.setDaemon(True)
+        self.event_thread.setName("MonitorHandler")
         self.event_thread.start()
 
     def init_sysinfo(self):
@@ -171,7 +172,6 @@ def init_sysinfo(self):
                                                  DISTRO_CODE_NAME,
                                                  platform.release())
         self.sysinfo.append(TelemetryEventParam("OSVersion", osversion))
-        self.sysinfo.append(TelemetryEventParam("GAVersion", CURRENT_AGENT))
         self.sysinfo.append(TelemetryEventParam("ExecutionMode", AGENT_EXECUTION_MODE))
 
         try:

tests/common/test_event.py

@@ -18,20 +18,23 @@
 from __future__ import print_function
 
 import json
+import os
+import threading
 from datetime import datetime, timedelta
 
+from mock import patch, Mock
+
+from azurelinuxagent.common import event, logger
 from azurelinuxagent.common.event import add_event, \
     WALAEventOperation, elapsed_milliseconds
 from azurelinuxagent.common.exception import EventError
 from azurelinuxagent.common.future import ustr
-from azurelinuxagent.common.utils.extensionprocessutil import read_output
 from azurelinuxagent.common.protocol.wire import GoalState
-from azurelinuxagent.common.version import CURRENT_VERSION
+from azurelinuxagent.common.utils import fileutil
+from azurelinuxagent.common.utils.extensionprocessutil import read_output
+from azurelinuxagent.common.version import CURRENT_VERSION, CURRENT_AGENT
 from azurelinuxagent.ga.monitor import MonitorHandler
-
-from tests.tools import *
-
-import azurelinuxagent.common.event as event
+from tests.tools import AgentTestCase, load_data, data_dir
 
 
 class TestEvent(AgentTestCase):
@@ -269,13 +272,29 @@ def test_periodic_emits_after_elapsed_delta(self, mock_event):
     @patch('azurelinuxagent.common.event.EventLogger.add_event')
     def test_periodic_forwards_args(self, mock_event):
         event.__event_logger__.reset_periodic()
+        event_time = datetime.utcnow().__str__()
+        event.add_periodic(logger.EVERY_DAY, "FauxEvent", op=WALAEventOperation.Log, is_success=True, duration=0,
+                           version=str(CURRENT_VERSION), message="FauxEventMessage", evt_type="", is_internal=False,
+                           log_event=True, force=False)
+        mock_event.assert_called_once_with("FauxEvent", op=WALAEventOperation.Log, is_success=True, duration=0,
+                                           version=str(CURRENT_VERSION), message="FauxEventMessage", evt_type="",
+                                           is_internal=False, log_event=True)
+
+    @patch("azurelinuxagent.common.event.datetime")
+    @patch('azurelinuxagent.common.event.EventLogger.add_event')
+    def test_periodic_forwards_args_default_values(self, mock_event, mock_datetime):
+        event.__event_logger__.reset_periodic()
+        event.add_periodic(logger.EVERY_DAY, "FauxEvent", message="FauxEventMessage")
+        mock_event.assert_called_once_with("FauxEvent", op=WALAEventOperation.Unknown, is_success=True, duration=0,
+                                           version=str(CURRENT_VERSION), message="FauxEventMessage", evt_type="",
+                                           is_internal=False, log_event=True)
 
-        event.add_periodic(logger.EVERY_DAY, "FauxEvent")
-        mock_event.assert_called_once_with(
-            "FauxEvent",
-            duration=0, evt_type='', is_internal=False, is_success=True,
-            log_event=True, message='', op=WALAEventOperation.Unknown,
-            version=str(CURRENT_VERSION))
+    @patch("azurelinuxagent.common.event.EventLogger.add_event")
+    def test_add_event_default_variables(self, mock_add_event):
+        add_event('test', message='test event')
+        mock_add_event.assert_called_once_with('test', duration=0, evt_type='', is_internal=False, is_success=True,
+                                               log_event=True, message='test event', op='Unknown',
+                                               version=str(CURRENT_VERSION))
 
     def test_save_event(self):
         add_event('test', message='test event')
@@ -296,19 +315,54 @@ def test_save_event_message_with_non_ascii_characters(self):
         duration = elapsed_milliseconds(datetime.utcnow())
         log_msg = "{0}\n{1}".format("DummyCmd", "\n".join([line for line in msg.split('\n') if line != ""]))
 
-        add_event('test_extension', message=log_msg, duration=duration)
+        with patch("azurelinuxagent.common.event.datetime") as patch_datetime:
+            patch_datetime.utcnow = Mock(return_value=datetime.strptime("2019-01-01 01:30:00",
+                                                                        '%Y-%m-%d %H:%M:%S'))
+            with patch('os.getpid', return_value=42):
+                with patch("threading.Thread.getName", return_value="HelloWorldTask"):
+                    add_event('test_extension', message=log_msg, duration=duration)
 
         for tld_file in os.listdir(self.tmp_dir):
             event_str = MonitorHandler.collect_event(os.path.join(self.tmp_dir, tld_file))
             event_json = json.loads(event_str)
 
-            self.assertEqual(len(event_json["parameters"]), 9)
+            self.assertEqual(len(event_json["parameters"]), 15)
 
+            # Checking the contents passed above, and also validating the default values that were passed in.
             for i in event_json["parameters"]:
                 if i["name"] == "Name":
                     self.assertEqual(i["value"], "test_extension")
-                if i["name"] == "Message":
+                elif i["name"] == "Message":
                     self.assertEqual(i["value"], log_msg)
+                elif i["name"] == "Version":
+                    self.assertEqual(i["value"], str(CURRENT_VERSION))
+                elif i['name'] == 'IsInternal':
+                    self.assertEqual(i['value'], False)
+                elif i['name'] == 'Operation':
+                    self.assertEqual(i['value'], 'Unknown')
+                elif i['name'] == 'OperationSuccess':
+                    self.assertEqual(i['value'], True)
+                elif i['name'] == 'Duration':
+                    self.assertEqual(i['value'], 0)
+                elif i['name'] == 'ExtensionType':
+                    self.assertEqual(i['value'], '')
+                elif i['name'] == 'ContainerId':
+                    self.assertEqual(i['value'], 'UNINITIALIZED')
+                elif i['name'] == 'OpcodeName':
+                    self.assertEqual(i['value'], '2019-01-01 01:30:00')
+                elif i['name'] == 'EventTid':
+                    self.assertEqual(i['value'], threading.current_thread().ident)
+                elif i['name'] == 'EventPid':
+                    self.assertEqual(i['value'], 42)
+                elif i['name'] == 'TaskName':
+                    self.assertEqual(i['value'], 'HelloWorldTask')
+                elif i['name'] == 'KeywordName':
+                    self.assertEqual(i['value'], '')
+                elif i['name'] == 'GAVersion':
+                    self.assertEqual(i['value'], str(CURRENT_AGENT))
+                else:
+                    self.assertFalse(True, "Contains a field outside the defaults expected. Field Name: {0}".
+                                     format(i['name']))
 
     def test_save_event_message_with_decode_errors(self):
         tmp_file = os.path.join(self.tmp_dir, "tmp_file")

tests/common/test_logger.py

@@ -18,11 +18,11 @@
 import json
 from datetime import datetime
 
-import azurelinuxagent.common.logger as logger
-from azurelinuxagent.common.event import add_log_event
-from azurelinuxagent.common.version import CURRENT_AGENT, CURRENT_VERSION
+from mock import patch, MagicMock
 
-from tests.tools import *
+from azurelinuxagent.common import logger
+from azurelinuxagent.common.event import add_log_event
+from tests.tools import AgentTestCase
 
 _MSG_INFO = "This is our test info logging message {0} {1}"
 _MSG_WARN = "This is our test warn logging message {0} {1}"
@@ -178,7 +178,7 @@ def test_telemetry_logger1(self, mock_save):
         self.assertEqual('FFF0196F-EE4C-4EAF-9AA5-776F622DEB4F', telemetry_json['providerId'])
         self.assertEqual(7, telemetry_json['eventId'])
 
-        self.assertEqual(5, len(telemetry_json['parameters']))
+        self.assertEqual(12, len(telemetry_json['parameters']))
         for x in telemetry_json['parameters']:
             if x['name'] == 'EventName':
                 self.assertEqual(x['value'], 'Log')