Merge remote-tracking branch 'origin/topic/jsiwek/broker'

* origin/topic/jsiwek/broker: (34 commits)
  Update broker submodule.
  Update broker submodule.
  broker integration: add missing baselines for doc tests
  broker integration: add prof.log statistics
  broker integration: add high-level usage documentation
  broker integration: add API documentation (broxygen/doxygen)
  broker integration: fix memory leak, add leak tests
  Update broker submodule.
  Improve comm tests.
  Fix gcc compile warnings.
  broker integration: fix unit tests to work when broker is not enabled.
  Add --enable-c++11 configure flag.
  broker integration: add (un)publish/(un)advertise functions
  broker integration: add knobs to set auto publish/advertise behavior
  broker integration: move listen port for unit tests to a btest variable
  broker integration: add events for incoming connection status updates
  broker integration: adapt to change in expiration_time
  Update coverage unit test baselines.
  broker integration: add Comm::enable function
  broker integration: process debug/diagnostic reports from broker
  ...

Conflicts:
	cmake
	testing/btest/Baseline/plugins.hooks/output

src/logging/Manager.cc

@@ -16,6 +16,10 @@
 #include "WriterBackend.h"
 #include "logging.bif.h"
 
+#ifdef ENABLE_BROKER
+#include "comm/Manager.h"
+#endif
+
 using namespace logging;
 
 struct Manager::Filter {
@@ -69,6 +73,11 @@ struct Manager::Stream {
 
 	WriterMap writers;	// Writers indexed by id/path pair.
 
+#ifdef ENABLE_BROKER
+	bool enable_remote;
+	int remote_flags;
+#endif
+
 	~Stream();
 	};
 
@@ -287,6 +296,11 @@ bool Manager::CreateStream(EnumVal* id, RecordVal* sval)
 	streams[idx]->event = event ? event_registry->Lookup(event->Name()) : 0;
 	streams[idx]->columns = columns->Ref()->AsRecordType();
 
+#ifdef ENABLE_BROKER
+	streams[idx]->enable_remote = internal_val("Log::enable_remote_logging")->AsBool();
+	streams[idx]->remote_flags = broker::PEERS;
+#endif
+
 	DBG_LOG(DBG_LOGGING, "Created new logging stream '%s', raising event %s",
 		streams[idx]->name.c_str(), event ? streams[idx]->event->Name() : "<none>");
 
@@ -828,6 +842,11 @@ bool Manager::Write(EnumVal* id, RecordVal* columns)
 #endif
 		}
 
+#ifdef ENABLE_BROKER
+	if ( stream->enable_remote )
+		comm_mgr->Log(id, columns, stream->remote_flags);
+#endif
+
 	Unref(columns);
 
 	if ( error )
@@ -1206,6 +1225,53 @@ void Manager::Terminate()
 		}
 	}
 
+#ifdef ENABLE_BROKER
+
+bool Manager::EnableRemoteLogs(EnumVal* stream_id, int flags)
+	{
+	auto stream = FindStream(stream_id);
+
+	if ( ! stream )
+		return false;
+
+	stream->enable_remote = true;
+	stream->remote_flags = flags;
+	return true;
+	}
+
+bool Manager::DisableRemoteLogs(EnumVal* stream_id)
+	{
+	auto stream = FindStream(stream_id);
+
+	if ( ! stream )
+		return false;
+
+	stream->enable_remote = false;
+	return true;
+	}
+
+bool Manager::RemoteLogsAreEnabled(EnumVal* stream_id)
+	{
+	auto stream = FindStream(stream_id);
+
+	if ( ! stream )
+		return false;
+
+	return stream->enable_remote;
+	}
+
+RecordType* Manager::StreamColumns(EnumVal* stream_id)
+	{
+	auto stream = FindStream(stream_id);
+
+	if ( ! stream )
+		return nullptr;
+
+	return stream->columns;
+	}
+
+#endif
+
 // Timer which on dispatching rotates the filter.
 class RotationTimer : public Timer {
 public:

src/logging/Manager.h

@@ -157,6 +157,34 @@ public:
 	 */
 	void Terminate();
 
+#ifdef ENABLE_BROKER
+	/**
+	 * Enable remote logs for a given stream.
+	 * @param stream_id the stream to enable remote logs for.
+	 * @param flags tune behavior of how log entries are sent to peer endpoints.
+	 * @return true if remote logs are enabled.
+	 */
+	bool EnableRemoteLogs(EnumVal* stream_id, int flags);
+
+	/**
+	 * Disable remote logs for a given stream.
+	 * @param stream_id the stream to disable remote logs for.
+	 * @return true if remote logs are disabled.
+	 */
+	bool DisableRemoteLogs(EnumVal* stream_id);
+
+	/**
+	 * @return true if remote logs are enabled for a given stream.
+	 */
+	bool RemoteLogsAreEnabled(EnumVal* stream_id);
+
+	/**
+	 * @return the type which corresponds to the columns in a log entry for
+	 * a given log stream.
+	 */
+	RecordType* StreamColumns(EnumVal* stream_id);
+#endif
+
 protected:
 	friend class WriterFrontend;
 	friend class RotationFinishedMessage;