Application.cc

// $Id$

/*************************************************************************
 * XDAQ Components for Distributed Data Acquisition                      *
 * Copyright (C) 2000-2019, CERN.                                        *
 * All rights reserved.                                                  *
 * Authors: L. Orsini, D. Simelevicius                                   *
 *                                                                       *
 * For the licensing terms see LICENSE.                                  *
 * For the list of contributors see CREDITS.                             *
 ************************************************************************/
#include "prometheus/Application.h"
#include "prometheus/counter.h"
#include "prometheus/summary.h"

#include <cstring>
#include <iterator>
#include <algorithm>
#include <vector>

#ifdef HAVE_ZLIB
#include <zlib.h>
#endif

#include "prometheus/serializer.h"
#include "prometheus/text_serializer.h"

#include "cgicc/CgiDefs.h"
#include "cgicc/Cgicc.h"
#include "cgicc/HTTPHTMLHeader.h"
#include "cgicc/HTMLClasses.h"
#include "cgicc/HTTPResponseHeader.h" 
#include "toolbox/task/TimerTask.h"
#include "toolbox/task/Timer.h"
#include "toolbox/task/TimerFactory.h"

#include "xdaq/ApplicationGroup.h" 
#include "xdaq/ApplicationRegistry.h"
#include "xdaq/EndpointAvailableEvent.h"

#include "xgi/framework/Method.h" 
#include "xcept/tools.h"


#include "toolbox/task/WorkLoopFactory.h"
#include "xdata/InfoSpaceFactory.h"
#include "xdata/exdr/Serializer.h"
#include "toolbox/Runtime.h"

#include "xdata/Table.h"
#include "xdata/exdr/Serializer.h"
#include "xdaq/ApplicationDescriptorImpl.h"

#include "toolbox/TimeVal.h"
#include "toolbox/stl.h"
#include "toolbox/utils.h"
#include "toolbox/regex.h"
#include "toolbox/BSem.h"
#include "toolbox/exception/Handler.h"

#include "xdata/exdr/Serializer.h"
#include "xdata/exdr/AutoSizeOutputStreamBuffer.h"
#include "xdata/exdr/FixedSizeInputStreamBuffer.h"
#include "xdata/exdr/FixedSizeOutputStreamBuffer.h"

#include <iostream>     // std::cout
#include <fstream>      // std::ifstream


static std::size_t WriteResponse(xgi::Input * in, xgi::Output * out, const std::string& body) 
{

	out->getHTTPResponseHeader().getStatusCode(200).getReasonPhrase(xgi::Utils::getResponsePhrase(200)).addHeader("Content-Type", "text/plain");

	#ifdef HAVE_ZLIB
	//auto acceptsGzip = IsEncodingAccepted(conn, "gzip");
	auto acceptsGzip = false;
	auto accept_encoding = in->getenv("ACCEPT_ENCODING");
	if (accept_encoding == "gzip" ) acceptsGzip = true;

	if (acceptsGzip) {
		auto compressed = GZipCompress(body);
		if (!compressed.empty()) {
			out->getHTTPResponseHeader().addHeader("Content-Encoding", "gzip");
			auto cl = toolbox::toString("%lu",static_cast<unsigned long>(compressed.size()));
			out->getHTTPResponseHeader().addHeader("Content-Length", cl.c_str());
			out->write(compressed.data(),  compressed.size());
			return compressed.size());
		}
	}
#endif

	auto bcl = toolbox::toString("%lu",static_cast<unsigned long>(body.size()));
	out->getHTTPResponseHeader().addHeader("Content-Length", bcl.c_str());

	out->write(body.data(), body.size());

	//DEBUG
	std::cout << body.data() << std::endl;

	return  body.size();
	//end write response
}


#ifdef HAVE_ZLIB
static std::vector<Byte> GZipCompress(const std::string& input) {
	auto zs = z_stream{};
	auto windowSize = 16 + MAX_WBITS;
	auto memoryLevel = 9;

	if (deflateInit2(&zs, Z_DEFAULT_COMPRESSION, Z_DEFLATED, windowSize,
			memoryLevel, Z_DEFAULT_STRATEGY) != Z_OK) {
		return {};
	}

	zs.next_in = (Bytef*)input.data();
	zs.avail_in = input.size();

	int ret;
	std::vector<Byte> output;
	output.reserve(input.size() / 2u);

	do {
		static const auto outputBytesPerRound = std::size_t{32768};

		zs.avail_out = outputBytesPerRound;
		output.resize(zs.total_out + zs.avail_out);
		zs.next_out = reinterpret_cast<Bytef*>(output.data() + zs.total_out);

		ret = deflate(&zs, Z_FINISH);

		output.resize(zs.total_out);
	} while (ret == Z_OK);

	deflateEnd(&zs);

	if (ret != Z_STREAM_END) {
		return {};
	}

	return output;
}
#endif

#include "xdata/Counter.h"

XDAQ_INSTANTIATOR_IMPL(prometheus::Application);


prometheus::Application::Application (xdaq::ApplicationStub* s) 
: xdaq::Application(s), xgi::framework::UIManager(this),
  exposer_registry_(std::make_shared<Registry>()),
  bytes_transferred_family_(BuildCounter().Name("exposer_transferred_bytes_total").Help("Transferred bytes to metrics services").Register(*exposer_registry_)),
  bytes_transferred_(bytes_transferred_family_.Add({})),
  num_scrapes_family_(BuildCounter().Name("exposer_scrapes_total") .Help("Number of times metrics were scraped") .Register(*exposer_registry_)),
  num_scrapes_(num_scrapes_family_.Add({})),
  request_latencies_family_(BuildSummary().Name("exposer_request_latencies") .Help("Latencies of serving scrape requests, in microseconds") .Register(*exposer_registry_)),
  request_latencies_(request_latencies_family_.Add( {}, Summary::Quantiles{{0.5, 0.05}, {0.9, 0.01}, {0.99, 0.001}})),
  user_exposer_registry_(std::make_shared<Registry>()),
  counter_family_(BuildCounter().Name("time_running_seconds_total").Help("How many seconds is this server running?").Labels({{"label", "value"}}).Register(*user_exposer_registry_)),
  second_counter_(counter_family_.Add( {{"another_label", "value"}, {"yet_another_label", "value"}}))
{


	//collectables_.push_back(exposer_registry_);
	//collectables_.push_back(user_exposer_registry_);
	residentcollectables_.push_back(exposer_registry_);
	residentcollectables_.push_back(user_exposer_registry_);

	s->getDescriptor()->setAttribute("icon", "/prometheus/images/prometheus_64x64.png");
	s->getDescriptor()->setAttribute("icon16x16", "/prometheus/images/prometheus_16x16.ico");

	//
	// bind HTTP callback
	xgi::bind(this, &prometheus::Application::metrics, "metrics");

	xgi::framework::deferredbind(this, this, &prometheus::Application::Default, "Default");

	// Bind setting of default parameters
	getApplicationInfoSpace()->addListener(this, "urn:xdaq-event:setDefaultValues");
	getApplicationContext()->addActionListener(this);

}

prometheus::Application::~Application()
{

}
void prometheus::Application::timeExpired(toolbox::task::TimerEvent& e)
{
	LOG4CPLUS_DEBUG (this->getApplicationLogger(), "Timer callback");

	//std::cout << "timer expired " << e.type()  << std::endl;
	if (e.getTimerTask()->name  == "urn:prometheus:check" )
	{

		std::vector<std::weak_ptr<Collectable>> ontheflycollectables;


		//std::cout << "going to process statistics" << std::endl;
		// user example update
		second_counter_.Increment();

		// scan infospaces and register items
		xdata::getInfoSpaceFactory()->lock();

		std::vector<xdata::InfoSpace*> spaces;

		auto iss = xdata::getInfoSpaceFactory()->match("urn:prometheus:(.*)");
		for (auto it = iss.begin(); it != iss.end(); ++it)
		{
			xdata::InfoSpace * is = dynamic_cast<xdata::InfoSpace *>((*it).second);
			is->lock();
			//std::cout << "---found infospace: " <<  is->name() << std::endl;
			LOG4CPLUS_DEBUG(this->getApplicationLogger(), "found infospace: " << is->name() );
			auto items = is->match("(.*)"); // get all items
			for (auto j = items.begin(); j != items.end(); j++)
			{
				if ( j->second->type().find("prometheus") != std::string::npos )
				{
					auto registry = dynamic_cast<xdata::Collectable*>(j->second)->registry_;


					///
					auto exists = [this,registry,j, ontheflycollectables]() {
						for (auto&& wcollectable : ontheflycollectables)
						{
							auto collectable = wcollectable.lock();
							if (!collectable) {
								continue;
							}

							if ( collectable == registry ) // not sure it compares the actual fired item
							{
								//std::cout << "---found already registered collectable: " << j->first << " of type: " << j->second->type() << std::endl;
								return true;
							}

						}
						return false;
					}();

					if (! exists )
					{
						//std::cout << "---register item: " << j->first <<  " of type: " <<  j->second->type() << std::endl;
						ontheflycollectables.push_back(registry);
					}


				}
			}
			is->unlock();
		}

		xdata::getInfoSpaceFactory()->unlock();
		// retrieved all available collectables and makes them available to prometheus
		mutex_.lock();
		collectables_.clear();
		std::copy(residentcollectables_.begin(), residentcollectables_.end(), back_inserter(collectables_));
		std::copy(ontheflycollectables.begin(), ontheflycollectables.end(), back_inserter(collectables_));
		mutex_.unlock();

		return;
	}


}

void prometheus::Application::actionPerformed( xdata::Event& event) 
{
	if (event.type() == "urn:xdaq-event:setDefaultValues")
	{	




	}
	else
	{
		LOG4CPLUS_ERROR (this->getApplicationLogger(), "Received unsupported event type '" << event.type() << "'");
	}
}


void prometheus::Application::actionPerformed(toolbox::Event& e) 
{
	LOG4CPLUS_DEBUG(this->getApplicationLogger(), "Received event " << e.type());
	if ( e.type() == "urn:xdaq-event:profile-loaded")
	{

		// activate a timer for user example
				toolbox::task::Timer * timer = 0;
				timer = toolbox::task::getTimerFactory()->createTimer("urn:prometheus:timer");

				toolbox::TimeInterval interval;
				toolbox::TimeVal start;
				start = toolbox::TimeVal::gettimeofday();

				interval.fromString("PT5S");
				timer->scheduleAtFixedRate( start, this,interval , 0, "urn:prometheus:check" );
	}
}

void prometheus::Application::metrics(xgi::Input * in, xgi::Output * out ) 
{
	cgicc::Cgicc cgi(in);
	auto start_time_of_request = std::chrono::steady_clock::now();

	//  collect metrics 
	auto collected_metrics = std::vector<MetricFamily>{};

	mutex_.lock();

	for (auto&& wcollectable : collectables_)
	{

		auto collectable = wcollectable.lock();
		if (!collectable) {
			continue;
		}

		auto&& metrics = collectable->Collect();
		//std::cout << "collectable insert" << std::endl;
		collected_metrics.insert(collected_metrics.end(), std::make_move_iterator(metrics.begin()), std::make_move_iterator(metrics.end()));
	}
	// end collect metrics
	mutex_.unlock();


	// serialize and output of metrics to requester 
	auto serializer = std::unique_ptr<Serializer>{new TextSerializer()};

	auto bodySize = WriteResponse(in, out, serializer->Serialize(collected_metrics));

	// update prometheus plugin metrics

	auto stop_time_of_request = std::chrono::steady_clock::now();
	auto duration = std::chrono::duration_cast<std::chrono::microseconds>( stop_time_of_request - start_time_of_request);

	request_latencies_.Observe(duration.count());
	bytes_transferred_.Increment(bodySize);
	num_scrapes_.Increment();

}

void prometheus::Application::Default(xgi::Input * in, xgi::Output * out ) 
{

	*out << cgicc::h3("XDAQ Prometheus Plugin") << std::endl;

}