AMBARI-17253. Ambari Alert causes too many wanings in ZooKeeper logs. (Masahiro Tanaka via swagle)

ambari-agent/src/main/python/ambari_agent/alerts/port_alert.py

@@ -38,6 +38,8 @@ class PortAlert(BaseAlert):
 
     self.uri = None
     self.default_port = None
+    self.socket_command = None
+    self.socket_command_response = None
     self.warning_timeout = DEFAULT_WARNING_TIMEOUT
     self.critical_timeout = DEFAULT_CRITICAL_TIMEOUT
 
@@ -61,6 +63,12 @@ class PortAlert(BaseAlert):
         'value' in reporting[reporting_state_critical]:
         self.critical_timeout = reporting[reporting_state_critical]['value']
 
+    if 'parameters' in alert_source_meta:
+      for parameter in alert_source_meta['parameters']:
+        if 'socket.command' == parameter['name']:
+          self.socket_command = parameter['value']
+        if 'socket.command.response' == parameter['name']:
+          self.socket_command_response = parameter['value']
 
     # check warning threshold for sanity
     if self.warning_timeout >= 30:
@@ -130,6 +138,12 @@ class PortAlert(BaseAlert):
 
       start_time = time.time()
       s.connect((host, port))
+      if self.socket_command is not None:
+        s.sendall(self.socket_command)
+        data = s.recv(1024)
+        if self.socket_command_response is not None and data != self.socket_command_response:
+          raise Exception("Expected response {0}, Actual response {1}".format(
+            self.socket_command_response, data))
       end_time = time.time()
       milliseconds = end_time - start_time
       seconds = milliseconds / 1000.0

ambari-agent/src/test/python/ambari_agent/TestPortAlert.py

@@ -363,3 +363,182 @@ class TestPortAlert(TestCase):
 
     alert.collect()
 
+  @patch("socket.socket")
+  @patch("time.time")
+  def test_collect_zookeeper(self, time, socket):
+    alert_meta = {
+      'name': 'zookeeper_server_process',
+      'label': 'label1',
+      'serviceName': 'service1',
+      'componentName': 'component1',
+      'uuid': '123',
+      'enabled': 'true'
+    }
+    alert_source_meta = {
+      'uri': 'http://192.168.0.1:2181',
+      'default_port': 80,
+      'parameters': [
+        {
+          'name': 'socket.command',
+          'display_name': 'Socket Command',
+          'value': 'ruok',
+          'description': 'test',
+          'type': 'STRING',
+          'visibility': 'HIDDEN'
+        },
+        {
+          'name': 'socket.command.response',
+          'display_name': 'Expected Response',
+          'value': 'imok',
+          'description': 'test',
+          'type': 'STRING',
+          'visibility': 'HIDDEN'
+        }
+      ]
+    }
+    cluster = 'c1'
+    host = 'host1'
+    expected_state = 'OK'
+    expected_text = 'TCP OK - 0.2010 response on port 2181'
+    time.side_effect = [123, 324, 567]
+    alert = PortAlert(alert_meta, alert_source_meta, self.config)
+    alert.set_cluster(cluster, host)
+    s = socket()
+    s.recv.return_value = "imok"
+
+    def collector_side_effect(clus, data):
+      self.assertEquals(data['name'], alert_meta['name'])
+      self.assertEquals(data['label'], alert_meta['label'])
+      self.assertEquals(data['service'], alert_meta['serviceName'])
+      self.assertEquals(data['component'], alert_meta['componentName'])
+      self.assertEquals(data['uuid'], alert_meta['uuid'])
+      self.assertEquals(data['enabled'], alert_meta['enabled'])
+      self.assertEquals(data['state'], expected_state)
+      self.assertEquals(data['text'], expected_text)
+      self.assertEquals(data['cluster'], cluster)
+      self.assertEquals(clus, cluster)
+
+    alert.collector = MagicMock()
+    alert.collector.put = Mock(side_effect=collector_side_effect)
+
+    alert.collect()
+
+  @patch("socket.socket")
+  @patch("time.time")
+  def test_collect_zookeeper_warning(self, time, socket):
+    alert_meta = {
+      'name': 'zookeeper_server_process',
+      'label': 'label1',
+      'serviceName': 'service1',
+      'componentName': 'component1',
+      'uuid': '123',
+      'enabled': 'true',
+    }
+    alert_source_meta = {
+      'uri': 'http://192.168.0.1:2181',
+      'default_port': 2181,
+      'parameters': [
+        {
+          'name': 'socket.command',
+          'display_name': 'Socket Command',
+          'value': 'ruok',
+          'description': 'test',
+          'type': 'STRING',
+          'visibility': 'HIDDEN'
+        },
+        {
+          'name': 'socket.command.response',
+          'display_name': 'Expected Response',
+          'value': 'imok',
+          'description': 'test',
+          'type': 'STRING',
+          'visibility': 'HIDDEN'
+        }
+      ]
+    }
+    cluster = 'c1'
+    host = 'host1'
+    expected_state = 'WARNING'
+    expected_text = 'TCP OK - 3.1170 response on port 2181'
+    time.side_effect = [123, 3240, 567]
+    alert = PortAlert(alert_meta, alert_source_meta, self.config)
+    alert.set_cluster(cluster, host)
+    s = socket()
+    s.recv.return_value = "imok"
+
+    def collector_side_effect(clus, data):
+      self.assertEquals(data['name'], alert_meta['name'])
+      self.assertEquals(data['label'], alert_meta['label'])
+      self.assertEquals(data['service'], alert_meta['serviceName'])
+      self.assertEquals(data['component'], alert_meta['componentName'])
+      self.assertEquals(data['uuid'], alert_meta['uuid'])
+      self.assertEquals(data['enabled'], alert_meta['enabled'])
+      self.assertEquals(data['state'], expected_state)
+      self.assertEquals(data['text'], expected_text)
+      self.assertEquals(data['cluster'], cluster)
+      self.assertEquals(clus, cluster)
+
+    alert.collector = MagicMock()
+    alert.collector.put = Mock(side_effect=collector_side_effect)
+
+    alert.collect()
+
+  @patch("socket.socket")
+  @patch("time.time")
+  def test_collect_zookeeper_connectionTimeout(self, time, socket):
+    alert_meta = {
+      'name': 'zookeeper_server_process',
+      'label': 'label1',
+      'serviceName': 'service1',
+      'componentName': 'component1',
+      'uuid': '123',
+      'enabled': 'true'
+    }
+    alert_source_meta = {
+      'uri': 'http://192.168.0.1:2181',
+      'default_port': 80,
+      'parameters': [
+        {
+          'name': 'socket.command',
+          'display_name': 'Socket Command',
+          'value': 'ruok',
+          'description': 'test',
+          'type': 'STRING',
+          'visibility': 'HIDDEN'
+        },
+        {
+          'name': 'socket.command.response',
+          'display_name': 'Expected Response',
+          'value': 'imok',
+          'description': 'test',
+          'type': 'STRING',
+          'visibility': 'HIDDEN'
+        }
+      ]
+    }
+    cluster = 'c1'
+    host = 'host1'
+    expected_state = 'CRITICAL'
+    expected_text = 'Connection failed: Socket Timeout to 192.168.0.1:2181'
+    time.side_effect = [123, 5240, 567]
+    alert = PortAlert(alert_meta, alert_source_meta, self.config)
+    alert.set_cluster(cluster, host)
+    s = socket()
+    s.recv.return_value = "imok"
+
+    def collector_side_effect(clus, data):
+      self.assertEquals(data['name'], alert_meta['name'])
+      self.assertEquals(data['label'], alert_meta['label'])
+      self.assertEquals(data['service'], alert_meta['serviceName'])
+      self.assertEquals(data['component'], alert_meta['componentName'])
+      self.assertEquals(data['uuid'], alert_meta['uuid'])
+      self.assertEquals(data['enabled'], alert_meta['enabled'])
+      self.assertEquals(data['state'], expected_state)
+      self.assertEquals(data['text'], expected_text)
+      self.assertEquals(data['cluster'], cluster)
+      self.assertEquals(clus, cluster)
+
+    alert.collector = MagicMock()
+    alert.collector.put = Mock(side_effect=collector_side_effect)
+
+    alert.collect()

ambari-server/src/main/java/org/apache/ambari/server/state/alert/PortSource.java

@@ -25,7 +25,7 @@ import com.google.gson.annotations.SerializedName;
  * Equality checking for instances of this class should be executed on every
  * member to ensure that reconciling stack differences is correct.
  */
-public class PortSource extends Source {
+public class PortSource extends ParameterizedSource {
 
   @SerializedName("uri")
   private String m_uri = null;

ambari-server/src/main/java/org/apache/ambari/server/upgrade/UpgradeCatalog240.java

@@ -698,6 +698,10 @@ public class UpgradeCatalog240 extends AbstractUpgradeCatalog {
       }});
       put("hive_metastore_process", defaultKeytabVisibilityMap);
       put("hive_server_process", defaultKeytabVisibilityMap);
+      put("zookeeper_server_process", new HashMap<String, String>(){{
+        put("socket.command", "HIDDEN");
+        put("socket.command.response", "HIDDEN");
+      }});
     }};
 
     Map<String, Map<String, String>> reportingPercentMap = new HashMap<String, Map<String, String>>(){{
@@ -832,6 +836,8 @@ public class UpgradeCatalog240 extends AbstractUpgradeCatalog {
               clusterID, "hive_webhcat_server_status");
       final AlertDefinitionEntity flumeAgentStatusAlertDefinitionEntity = alertDefinitionDAO.findByName(
               clusterID, "flume_agent_status");
+      final AlertDefinitionEntity zookeeperServerProcessAlertDefinitionEntity = alertDefinitionDAO.findByName(
+              clusterID, "zookeeper_server_process");
 
       Map<AlertDefinitionEntity, List<String>> alertDefinitionParams = new HashMap<>();
       checkedPutToMap(alertDefinitionParams, namenodeLastCheckpointAlertDefinitionEntity,
@@ -852,6 +858,9 @@ public class UpgradeCatalog240 extends AbstractUpgradeCatalog {
               Lists.newArrayList("default.smoke.user", "connection.timeout"));
       checkedPutToMap(alertDefinitionParams, flumeAgentStatusAlertDefinitionEntity,
               Lists.newArrayList("run.directory"));
+      checkedPutToMap(alertDefinitionParams, zookeeperServerProcessAlertDefinitionEntity,
+              Lists.newArrayList("socket.command", "socket.command.response"));
+
 
       Map<Long, AlertDefinitionEntity> definitionsForPropertyUpdates = new HashMap<>();
 
@@ -1315,6 +1324,28 @@ public class UpgradeCatalog240 extends AbstractUpgradeCatalog {
       paramsToAdd.add(param);
 
     }
+    if (params.contains("socket.command")) {
+      JsonObject param = new JsonObject();
+      param.add("name", new JsonPrimitive("socket.command"));
+      param.add("display_name", new JsonPrimitive("Socket Command"));
+      param.add("value", new JsonPrimitive("ruok"));
+      param.add("type", new JsonPrimitive("STRING"));
+      param.add("description", new JsonPrimitive("A socket command which queries ZooKeeper to respond with its state. The expected response is imok."));
+      paramsToAdd.add(param);
+
+    }
+    if (params.contains("socket.command.response")) {
+      JsonObject param = new JsonObject();
+      param.add("name", new JsonPrimitive("socket.command.response"));
+      param.add("display_name", new JsonPrimitive("Expected Response"));
+      param.add("value", new JsonPrimitive("imok"));
+      param.add("type", new JsonPrimitive("STRING"));
+      param.add("description", new JsonPrimitive("The expected response to the socket command."));
+      paramsToAdd.add(param);
+
+    }
+
+
 
     if (!parameterExists) {
       parametersJson = new JsonArray();

ambari-server/src/main/resources/common-services/ZOOKEEPER/3.4.5/alerts.json

@@ -52,7 +52,25 @@
               "text": "Connection failed: {0} to {1}:{2}",
               "value": 5.0
             }
-          }
+          },
+          "parameters": [
+            {
+              "name": "socket.command",
+              "display_name": "Socket Command",
+              "value": "ruok",
+              "description": "A socket command which queries ZooKeeper to respond with its state. The expected response is imok.",
+              "type": "STRING",
+              "visibility": "HIDDEN"
+            },
+            {
+              "name": "socket.command.response",
+              "display_name": "Expected Response",
+              "value": "imok",
+              "description": "The expected response to the socket command.",
+              "type": "STRING",
+              "visibility": "HIDDEN"
+            }
+          ]
         }
       }
     ]