FOGL-9963 Add Alerts as a source for notifications

C/services/notification/data_availability_rule.cpp

@@ -45,6 +45,13 @@ static const char *default_config = QUOTE({
 		"default" : "",
 		"displayName" : "Asset Code",
 		"order" : "3"
+	},
+	"alerts" : {
+		"description" : "Deliver alert data to the notification delivery mechanism",
+		"type" : "boolean",
+		"default" : "false",
+		"displayName" : "Alerts",
+		"order" : "4"
 	}
 });
 
@@ -147,6 +154,12 @@ string DataAvailabilityRule::triggers()
 		ret += "{ \"audit\" : \"" + audit + "\" }";
 		comma = ",";
 	}
+	if (m_alert)
+	{
+		ret += comma;
+		ret += "{ \"alert\" : \"alert\" }";
+		comma = ",";
+	}
 	ret += " ] }";
 	return ret;
 }
@@ -339,5 +352,14 @@ void DataAvailabilityRule::configure(const ConfigCategory &config)
 		DatapointValue value (m_assetCodeList[i]);
 		handle->addTrigger(m_assetCodeList[i], new RuleTrigger(m_assetCodeList[i], new Datapoint(m_assetCodeList[i], value)));
 	}
+
+	string alerts = config.getValue("alerts");
+	m_alert = alerts[0] == 't' ? true : false;
+	if (m_alert)
+	{
+		DatapointValue dpv("alert");
+		handle->addTrigger("alert", new RuleTrigger("alert", new Datapoint("alert", dpv)));
+	}
 
+
 }

C/services/notification/delivery_plugin.cpp

@@ -9,8 +9,11 @@
  */
 
 #include <delivery_plugin.h>
+#include <rapidjson/document.h>
+#include <rapidjson/writer.h>
 
 using namespace std;
+using namespace rapidjson;
 
 
 // DeliveryPlugin constructor
@@ -115,7 +118,7 @@ bool DeliveryPlugin::deliver(const std::string& deliveryName,
 					     deliveryName,
 					     notificationName,
 					     triggerReason,
-					     message);
+					     expandMacros(message, triggerReason));
 	}
 	unsigned int duration = time(0) - start;
 	if (duration > 5)
@@ -188,3 +191,142 @@ void DeliveryPlugin::registerService(void *func, void *data)
 		(*pluginRegisterService)(m_instance, func, data);
 	}
 }
+
+
+/**
+ * Create the information about the macros to substitute in the given string
+ *
+ * @param str	The string we are substituting
+ * @param macros	Vector of macros to build up
+ */
+void DeliveryPlugin::collectMacroInfo(const string& str, vector<Macro>& macros)
+{
+	string::size_type start = str.find('$');
+	string::size_type end = str.find('$', start + 1);
+
+	while (start != string::npos && end != string::npos) 
+	{
+		string::size_type bar = str.find('|', start + 1);
+		if (bar != string::npos && bar < end && bar > start + 1) 
+		{
+			string def = str.substr(bar + 1, end - bar - 1);
+			macros.emplace_back(Macro(str.substr(start + 1, bar - start - 1), start, def));
+		}
+		else if (end > start + 1)
+		{
+			macros.emplace_back(Macro(str.substr(start + 1, end - start - 1), start));
+		}
+		start = str.find('$', end + 1);
+		end = str.find('$', start + 1);
+	}
+}
+
+/**
+ * Substitute values from the reason into the string.
+ * Macros are of the form $key$ or 
+ * $key|default$. Where key is one of the keys found in
+ * the notification data of the reason document. Keys
+ * are typical datapoint names that triggered the alert
+ * for readings, statis values for statistics data and
+ * messages for alert data.
+ *
+ * @param message	The string to substitute into
+ * @param reason	The notification reason from which to pull data
+ * @return string	The substituted string
+ */
+string  DeliveryPlugin::expandMacros(const string& message, const string& reason)
+{
+	string rval = message;
+	vector<Macro> macros;
+	Logger::getLogger()->debug("Expand macros in message %s with reason %s",
+			message.c_str(), reason.c_str());
+	collectMacroInfo(rval, macros);
+	if (macros.size())
+	{
+		Document doc;
+		doc.Parse(reason.c_str());
+		if (doc.HasParseError())
+		{
+			// Failed to parse the reason, ignore macros
+			Logger::getLogger()->warn("Unable to parse reason document, macro substitutios within the notification will be ignored. The reason document is:  %s", reason.c_str());
+			return rval;
+		}
+		if (!doc.HasMember("data"))
+		{
+			Logger::getLogger()->warn("Unable to perform macro substitution in the notifcation alert. No data element was found in reason document %s", reason.c_str());
+			return rval;
+		}
+		Value& data = doc["data"];
+		Value::ConstMemberIterator itr = data.MemberBegin();
+		if (itr == data.MemberEnd())
+		{
+			Logger::getLogger()->warn("Unable to perform macro substitution in the notifcation alert. Data element has no children, reason document %s", reason.c_str());
+			return rval;
+		}
+		const Value& v = itr->value;
+		string assetName = itr->name.GetString();
+
+
+		// Replace Macros by datapoint value
+		for (auto it =  macros.rbegin(); it != macros.rend(); ++it)
+		{
+			if (it->name == "ASSET")
+			{
+				rval.replace(it->start, it->name.length()+2
+							+ (it->def.empty() ? 0 : it->def.length() + 1),
+							assetName);
+			}
+			else if (v.HasMember(it->name.c_str()))
+			{
+				string val;
+				if (v[it->name.c_str()].IsString())
+				{
+					val = v[it->name.c_str()].GetString();
+				}
+				else if (v[it->name.c_str()].IsInt64())
+				{
+					val = to_string(v[it->name.c_str()].GetInt64());
+				}
+				else if (v[it->name.c_str()].IsDouble())
+				{
+					val = to_string(v[it->name.c_str()].GetDouble());
+					// Trim trailing 0's
+					size_t len = val.length();
+					while (len > 0 && val[len-1] == '0')
+					{
+						len--;
+					}
+					if (len > 0)
+					{
+						val = val.substr(0, len);
+					}
+
+				}
+				else if (v[it->name.c_str()].IsObject())
+				{
+					StringBuffer strbuf;
+					Writer<rapidjson::StringBuffer> writer(strbuf);
+					v[it->name.c_str()].Accept(writer);
+			                val = strbuf.GetString();
+				}
+				else
+				{
+					Logger::getLogger()->warn("The datapoint %s cannot be used as a macro substitution as it is not a string, numeric value or JSON document",it->name.c_str());
+					continue;
+				}
+				rval.replace(it->start, it->name.length()+2
+							+ (it->def.empty() ? 0 : it->def.length() + 1),
+							val);
+			}
+			else if (!it->def.empty())
+			{
+				rval.replace(it->start, it->name.length() + it->def.length() + 3, it->def);
+			}
+			else
+			{
+				rval.replace(it->start, it->name.length()+2, "");
+			}
+		}
+	}
+	return rval;
+}

C/services/notification/delivery_queue.cpp

@@ -24,6 +24,7 @@
 #include <notification_queue.h>
 #include <delivery_queue.h>
 
+
 using namespace std;
 
 DeliveryQueue* DeliveryQueue::m_instance = 0;
@@ -436,10 +437,11 @@ void DeliveryQueue::processDelivery(DeliveryQueueElement* elem)
 	{
 		// Call plugin_deliver
 		std::string reason = elem->getData()->getReason();
+		std::string message = elem->getData()->getMessage();
 		bool deliverSuccessFlag =  elem->getPlugin()->deliver(elem->getName(),
 					   elem->getData()->getNotificationName(),
 					   reason,
-					   elem->getData()->getMessage());
+					   message);
 
 		std::string instanceName;
                 const NotificationInstance* nInstance = elem->getData()->getInstance();
@@ -470,3 +472,4 @@ void DeliveryQueue::processDelivery(DeliveryQueueElement* elem)
 	}
 #endif
 }
+

C/services/notification/include/data_availability_rule.h

@@ -39,6 +39,7 @@ class DataAvailabilityRule : public RulePlugin
 private:
 	std::vector<std::string>	m_assetCodeList;
 	std::vector<std::string>	m_auditCodeList;
+	bool				m_alert;
 };
 
 #endif

C/services/notification/include/delivery_plugin.h

@@ -41,6 +41,7 @@ class DeliveryPlugin : public Plugin
 		void			registerIngest(void *func, void *data);
 		void			registerService(void *func, void *data);
 		bool			isEnabled() { return m_enabled; };
+		std::string		expandMacros(const std::string& message, const std::string& reason);
 
 	private:
 		PLUGIN_HANDLE		(*pluginInit)(const ConfigCategory* config);
@@ -54,6 +55,27 @@ class DeliveryPlugin : public Plugin
 							        const std::string& newConfig);
 		void			(*pluginStartPtr)(PLUGIN_HANDLE);
 		void			setEnabled(const ConfigCategory& config);
+		class Macro {
+			public:
+				Macro(const std::string& dpname, std::string::size_type s,
+						const std::string& defValue) :
+					start(s), name(dpname), def(defValue)
+
+				{
+				};
+				Macro(const std::string& dpname, std::string::size_type s) :
+					start(s), name(dpname)
+
+				{
+				};
+				// Start of variable to substitute
+				std::string::size_type		start;
+				// Name of variable to substitute
+				std::string			name;
+				// Default value to substitute
+				std::string			def;
+		};
+		void			collectMacroInfo(const std::string& str, std::vector<Macro>& macros);
 
 	public:
 		// Persist plugin data

C/services/notification/include/delivery_queue.h

@@ -43,7 +43,6 @@ class DeliveryDataElement
 					getInstance() { return m_instance; };
 		NotificationInstance*
 					m_instance;
-
 	private:
 		DeliveryPlugin* m_plugin;
 		std::string	m_deliveryName;
@@ -125,6 +124,7 @@ class DeliveryQueue
 	private:
 		void			processDelivery(DeliveryQueueElement* data);
 
+
 	private:
 
 		const std::string	m_name;

C/services/notification/include/notification_api.h

@@ -24,6 +24,7 @@ using HttpServer = SimpleWeb::Server<SimpleWeb::HTTP>;
 #define RECEIVE_AUDIT_NOTIFICATION	"^/notification/reading/audit/([A-Za-z][a-zA-Z0-9_%\\-\\.]*)$"
 #define RECEIVE_STATS_NOTIFICATION	"^/notification/reading/stat/([A-Za-z0-9][a-zA-Z0-9_%\\-\\.]*)$"
 #define RECEIVE_STATS_RATE_NOTIFICATION	"^/notification/reading/rate/([A-Za-z0-9][a-zA-Z0-9_%\\-\\.]*)$"
+#define RECEIVE_ALERT_NOTIFICATION	"^/notification/reading/alert$"
 #define GET_NOTIFICATION_INSTANCES	"^/notification$"
 #define GET_NOTIFICATION_DELIVERY	"^/notification/delivery$"
 #define GET_NOTIFICATION_RULES		"^/notification/rules$"
@@ -81,6 +82,8 @@ class NotificationApi
 						shared_ptr<HttpServer::Request> request);
 		void		processStatsRateCallback(shared_ptr<HttpServer::Response> response,
 						shared_ptr<HttpServer::Request> request);
+		void		processAlertCallback(shared_ptr<HttpServer::Response> response,
+						shared_ptr<HttpServer::Request> request);
 		void		getNotificationObject(NOTIFICATION_OBJECT object,
 						      shared_ptr<HttpServer::Response> response,
 						      shared_ptr<HttpServer::Request> request);
@@ -97,6 +100,8 @@ class NotificationApi
 				getStatsCallbackURL() const { return m_statsCallbackURL; };
 		const std::string&
 				getStatsRateCallbackURL() const { return m_statsRateCallbackURL; };
+		const std::string&
+				getAlertCallbackURL() const { return m_alertCallbackURL; };
 		void		setCallBackURL();
 		bool		removeNotification(const std::string& notificationName);
 		// Add asset name and data to the Readings process queue
@@ -108,6 +113,7 @@ class NotificationApi
 						  const string& payload);
 		bool		queueStatsRateNotification(const string& auditCode,
 						  const string& payload);
+		bool		queueAlertNotification(const string& payload);
 
 		void		defaultResource(shared_ptr<HttpServer::Response> response,
                                         shared_ptr<HttpServer::Request> request);
@@ -133,6 +139,7 @@ class NotificationApi
 		std::string			m_auditCallbackURL;
 		std::string			m_statsCallbackURL;
 		std::string			m_statsRateCallbackURL;
+		std::string			m_alertCallbackURL;
 		Logger*				m_logger;
 };
 

C/services/notification/include/notification_subscription.h

@@ -136,6 +136,23 @@ class StatsRateSubscriptionElement : public SubscriptionElement
 		std::string	m_stat;
 };
 
+/**
+ * The SubscriptionElement class handles the notification registration to
+ * storage server based on alerts and its notification name.
+ */
+class AlertSubscriptionElement : public SubscriptionElement
+{
+	public:
+		AlertSubscriptionElement(const std::string& notificationName,
+				    NotificationInstance* notification);
+
+		~AlertSubscriptionElement();
+
+		bool		registerSubscription(StorageClient& storage) const;
+		bool		unregister(StorageClient& storage) const;
+		string		getKey() const { return string("alert::alert"); };
+};
+
 /**
  * The NotificationSubscription class handles all notification registrations to
  * storage server.