unresolved relations

agent-ovs/lib/PrometheusManager.cpp

@@ -140,6 +140,7 @@ static string ofpeer_family_names[] =
   "opflex_peer_ep_undeclare_err_count",
   "opflex_peer_state_report_req_count",
   "opflex_peer_state_report_resp_count",
+  "opflex_peer_unresolved_policy_count",
   "opflex_peer_state_report_err_count"
 };
 
@@ -163,6 +164,7 @@ static string ofpeer_family_help[] =
   "number of endpoint undeclare error responses from opflex peer",
   "number of state reports sent to opflex peer",
   "number of state reports responses received from opflex peer",
+  "number of unresolved policies from the opflex peer",
   "number of state reports error repsonses from opflex peer"
 };
 
@@ -2933,6 +2935,7 @@ void PrometheusManager::addNUpdateOFPeerStats (void)
                             SysStatUniverse::resolve(framework);
     std::unordered_map<string, OF_SHARED_PTR<OFStats>> stats;
     agent.getFramework().getOpflexPeerStats(stats);
+    int unresolved_pol_count = agent.getFramework().getUnresolvedItem();
     if (su) {
         for (const auto& peerStat : stats) {
             optional<shared_ptr<OpflexCounter>> counter =
@@ -3007,6 +3010,9 @@ void PrometheusManager::addNUpdateOFPeerStats (void)
                 case OFPEER_STATE_REPORT_RESPS:
                     metric_opt = counter.get()->getStateReportResps();
                     break;
+                case OFPEER_UNRESOLVED_POL:
+                    metric_opt = unresolved_pol_count;
+                    break;
                 case OFPEER_STATE_REPORT_ERRS:
                     metric_opt = counter.get()->getStateReportErrs();
                     break;

agent-ovs/lib/include/opflexagent/PrometheusManager.h

@@ -730,6 +730,7 @@ class PrometheusManager {
         OFPEER_EP_UNDECLARE_ERRS,
         OFPEER_STATE_REPORTS,
         OFPEER_STATE_REPORT_RESPS,
+        OFPEER_UNRESOLVED_POL,
         OFPEER_STATE_REPORT_ERRS,
         OFPEER_METRICS_MAX = OFPEER_STATE_REPORT_ERRS
     };

libopflex/engine/OpflexPEHandler.cpp

@@ -365,12 +365,14 @@ void OpflexPEHandler::handlePolicyResolveRes(uint64_t reqId,
                        << "Malformed policy resolve response: policy must be array";
             conn->disconnect();
         }
-
         Value::ConstValueIterator it;
         for (it = policy.Begin(); it != policy.End(); ++it) {
             const Value& mo = *it;
             serializer.deserialize(mo, *client, true, &notifs);
-        }
+            const Value& uriv = mo["uri"];
+            getProcessor()->removePendingItem(conn->getHostname(), uriv.GetString()); 
+
+      }
     }
     client->deliverNotifications(notifs);
 }

libopflex/engine/OpflexPool.cpp

@@ -426,6 +426,7 @@ size_t OpflexPool::sendToRole(OpflexMessage* message,
         }
         incrementMsgCounter(conn, m_copy);
         conn->sendMessage(m_copy, sync);
+        setPeername(conn->getHostname(), conn->getPort());
         i += 1;
     }
     // all allocated buffers should have been dispatched to

libopflex/engine/Processor.cpp

@@ -20,6 +20,8 @@
 #include <limits>
 #include <cmath>
 #include <random>
+#include <fstream>
+#include <unordered_map>
 
 #include <boost/tuple/tuple.hpp>
 #include <boost/foreach.hpp>
@@ -80,6 +82,25 @@ inline uint64_t now(uv_loop_t* loop) {
     return uv_now(loop);
 }
 
+void Processor::setPendingItem(std::pair<std::string, int> peerName, std::string uri) {
+    std::unordered_map<std::string, std::string>::iterator it;
+    pendingResolution[peerName.first] = uri;
+    for(it = pendingResolution.begin(); it != pendingResolution.end(); it++){
+}
+}
+
+void Processor::removePendingItem(std::string hostname, std::string uri) {
+     std::unordered_map<std::string, std::string>::iterator it;
+     for(it = pendingResolution.begin(); it != pendingResolution.end(); it++){
+         if (it->first == hostname && it->second == uri) {
+             pendingResolution.erase(it);
+         }
+     }
+     for(it = pendingResolution.begin(); it != pendingResolution.end(); it++){
+     }
+     unResolvedItem = pendingResolution.size();
+     }
+
 Processor::change_expiration::change_expiration(uint64_t new_exp_)
     : new_exp(new_exp_) {}
 
@@ -279,6 +300,8 @@ bool Processor::resolveObj(ClassInfo::class_type_t type, const item& i,
             PolicyResolveReq* req =
                 new PolicyResolveReq(this, nextXid++, refs);
             sendToRole(i, newexp, req, OFConstants::POLICY_REPOSITORY);
+            std::pair<std::string, int> peerName = pool.getPeername();
+            setPendingItem(peerName, i.uri.toString());
             return true;
         }
         break;
@@ -374,7 +397,6 @@ void Processor::processItem(obj_state_by_exp::iterator& it) {
                << " of class " << ci.getName()
                << " and type " << ci.getType()
                << " in state " << curState;
-
     ItemState newState;
 
     switch (curState) {

libopflex/engine/include/opflex/engine/Processor.h

@@ -209,6 +209,18 @@ class Processor : public internal::AbstractObjectListener,
      */
     uint64_t getPrrTimerDuration() { return prrTimerDuration; }
 
+    std::unordered_map<std::string, std::string> pendingResolution; 
+
+    int unResolvedItem;
+
+    void setPendingItem(std::pair<std::string, int> peerName, std::string uri); 
+
+    void removePendingItem(std::string hostname, std::string uri); 
+
+    int getUnresolvedItem(){
+        return unResolvedItem;
+    }
+
     // See HandlerFactory::newHandler
     virtual
     internal::OpflexHandler* newHandler(internal::OpflexConnection* conn);

libopflex/engine/include/opflex/engine/internal/OpflexPool.h

@@ -335,6 +335,14 @@ class OpflexPool : private boost::noncopyable {
     opflex::modb::MAC getTunnelMac() {
         return tunnelMac;
     }
+
+    void setPeername(const std::string hostname, int port) {
+       peerName = std::make_pair(hostname,port); 
+    }
+
+    std::pair<std::string, int> getPeername(){
+       return peerName;
+    }
 
     /**
      * Retrieve OpFlex client stats for each available peer
@@ -391,6 +399,7 @@ class OpflexPool : private boost::noncopyable {
     boost::asio::ip::address_v4 ipv6_proxy;
     boost::asio::ip::address_v4 mac_proxy;
     opflex::modb::MAC tunnelMac;
+    std::pair<std::string, int> peerName;
 
     uv_loop_t* client_loop;
     uv_async_t conn_async;

libopflex/include/opflex/ofcore/OFFramework.h

@@ -729,6 +729,7 @@ class OFFramework : private boost::noncopyable {
      */
     void setTunnelMac(const opflex::modb::MAC &mac);
 
+    int getUnresolvedItem(); 
     /**
      * set the prr (policy resolve request) timer durarion.
      * @param duration timer duration in milliseconds

libopflex/ofcore/OFFramework.cpp

@@ -281,6 +281,11 @@ void OFFramework::getOpflexPeerStats(std::unordered_map<string, OF_SHARED_PTR<OF
     pool.getOpflexPeerStats(stats);
 }
 
+int OFFramework::getUnresolvedItem() {
+    int countUnresolved = pimpl->processor.getUnresolvedItem();
+    return countUnresolved ;
+}
+
 void OFFramework::overrideObservableReporting(modb::class_id_t class_id, bool enabled) {
     pimpl->processor.overrideObservableReporting(class_id, enabled);
 }