[thirdparty/pdns.git] / pdns / dnsdist.hh

#pragma once
#include "config.h"
#include "ext/luawrapper/include/LuaContext.hpp"
#include <time.h>
#include "misc.hh"
#include "iputils.hh"
#include "dnsname.hh"
#include <atomic>
#include <boost/circular_buffer.hpp>
#include <boost/variant.hpp>
#include <mutex>
#include <thread>
#include "sholder.hh"
#include "dnscrypt.hh"
#include "dnsdist-cache.hh"

void* carbonDumpThread();
uint64_t uptimeOfProcess(const std::string& str);

struct DynBlock
{
  DynBlock& operator=(const DynBlock& rhs)
  {
    reason=rhs.reason;
    until=rhs.until;
    blocks.store(rhs.blocks);
    return *this;
  }
  
  string reason;
  struct timespec until;
  mutable std::atomic<unsigned int> blocks;
};

extern GlobalStateHolder<NetmaskTree<DynBlock>> g_dynblockNMG;

extern vector<pair<struct timeval, std::string> > g_confDelta;

struct DNSDistStats
{
  using stat_t=std::atomic<uint64_t>; // aww yiss ;-)
  stat_t responses{0};
  stat_t servfailResponses{0};
  stat_t queries{0};
  stat_t nonCompliantQueries{0};
  stat_t nonCompliantResponses{0};
  stat_t rdQueries{0};
  stat_t emptyQueries{0};
  stat_t aclDrops{0};
  stat_t blockFilter{0};
  stat_t dynBlocked{0};
  stat_t ruleDrop{0};
  stat_t ruleNXDomain{0};
  stat_t selfAnswered{0};
  stat_t downstreamTimeouts{0};
  stat_t downstreamSendErrors{0};
  stat_t truncFail{0};
  stat_t noPolicy{0};
  stat_t cacheHits{0};
  stat_t cacheMisses{0};
  stat_t latency0_1{0}, latency1_10{0}, latency10_50{0}, latency50_100{0}, latency100_1000{0}, latencySlow{0};
  
  double latencyAvg100{0}, latencyAvg1000{0}, latencyAvg10000{0}, latencyAvg1000000{0};
  typedef std::function<uint64_t(const std::string&)> statfunction_t;
  typedef boost::variant<stat_t*, double*, statfunction_t> entry_t;
  std::vector<std::pair<std::string, entry_t>> entries{
    {"responses", &responses}, {"servfail-responses", &servfailResponses},
    {"queries", &queries}, {"acl-drops", &aclDrops},
    {"block-filter", &blockFilter}, {"rule-drop", &ruleDrop},
    {"rule-nxdomain", &ruleNXDomain}, {"self-answered", &selfAnswered},
    {"downstream-timeouts", &downstreamTimeouts}, {"downstream-send-errors", &downstreamSendErrors}, 
    {"trunc-failures", &truncFail}, {"no-policy", &noPolicy},
    {"latency0-1", &latency0_1}, {"latency1-10", &latency1_10},
    {"latency10-50", &latency10_50}, {"latency50-100", &latency50_100}, 
    {"latency100-1000", &latency100_1000}, {"latency-slow", &latencySlow},
    {"latency-avg100", &latencyAvg100}, {"latency-avg1000", &latencyAvg1000}, 
    {"latency-avg10000", &latencyAvg10000}, {"latency-avg1000000", &latencyAvg1000000},
    {"uptime", uptimeOfProcess},
    {"real-memory-usage", getRealMemoryUsage},
    {"noncompliant-queries", &nonCompliantQueries},
    {"noncompliant-responses", &nonCompliantResponses},
    {"rdqueries", &rdQueries},
    {"empty-queries", &emptyQueries},
    {"cache-hits", &cacheHits},
    {"cache-misses", &cacheMisses},
    {"cpu-user-msec", getCPUTimeUser},
    {"cpu-sys-msec", getCPUTimeSystem},
    {"fd-usage", getOpenFileDescriptors}, {"dyn-blocked", &dynBlocked}, 
    {"dyn-block-nmg-size", [](const std::string&) { return g_dynblockNMG.getLocal()->size(); }}
  };
};


extern struct DNSDistStats g_stats;


struct StopWatch
{
#ifndef CLOCK_MONOTONIC_RAW
#define CLOCK_MONOTONIC_RAW CLOCK_MONOTONIC
#endif
  struct timespec d_start{0,0};
  void start() {  
    if(clock_gettime(CLOCK_MONOTONIC_RAW, &d_start) < 0)
      unixDie("Getting timestamp");
    
  }
  
  double udiff() const {
    struct timespec now;
    if(clock_gettime(CLOCK_MONOTONIC_RAW, &now) < 0)
      unixDie("Getting timestamp");
    
    return 1000000.0*(now.tv_sec - d_start.tv_sec) + (now.tv_nsec - d_start.tv_nsec)/1000.0;
  }

  double udiffAndSet() {
    struct timespec now;
    if(clock_gettime(CLOCK_MONOTONIC_RAW, &now) < 0)
      unixDie("Getting timestamp");
    
    auto ret= 1000000.0*(now.tv_sec - d_start.tv_sec) + (now.tv_nsec - d_start.tv_nsec)/1000.0;
    d_start = now;
    return ret;
  }

};

class QPSLimiter
{
public:
  QPSLimiter()
  {
  }

  QPSLimiter(unsigned int rate, unsigned int burst) : d_rate(rate), d_burst(burst), d_tokens(burst)
  {
    d_passthrough=false;
    d_prev.start();
  }

  unsigned int getRate() const
  {
    return d_passthrough? 0 : d_rate;
  }

  int getPassed() const
  {
    return d_passed;
  }
  int getBlocked() const
  {
    return d_blocked;
  }

  bool check() const // this is not quite fair
  {
    if(d_passthrough)
      return true;
    auto delta = d_prev.udiffAndSet();
  
    d_tokens += 1.0*d_rate * (delta/1000000.0);

    if(d_tokens > d_burst)
      d_tokens = d_burst;

    bool ret=false;
    if(d_tokens >= 1.0) { // we need this because burst=1 is weird otherwise
      ret=true;
      --d_tokens;
      d_passed++;
    }
    else
      d_blocked++;

    return ret; 
  }
private:
  bool d_passthrough{true};
  unsigned int d_rate;
  unsigned int d_burst;
  mutable double d_tokens;
  mutable StopWatch d_prev;
  mutable unsigned int d_passed{0};
  mutable unsigned int d_blocked{0};
};

struct IDState
{
  IDState() : origFD(-1), delayMsec(0) { origDest.sin4.sin_family = 0;}
  IDState(const IDState& orig)
  {
    origFD = orig.origFD;
    origID = orig.origID;
    origRemote = orig.origRemote;
    origDest = orig.origDest;
    delayMsec = orig.delayMsec;
    age.store(orig.age.load());
  }

  int origFD;  // set to <0 to indicate this state is empty   // 4

  ComboAddress origRemote;                                    // 28
  ComboAddress origDest;                                      // 28
  StopWatch sentTime;                                         // 16
  DNSName qname;                                              // 80
#ifdef HAVE_DNSCRYPT
  std::shared_ptr<DnsCryptQuery> dnsCryptQuery{0};
#endif
  std::shared_ptr<DNSDistPacketCache> packetCache{nullptr};
  uint32_t cacheKey;                                          // 8
  std::atomic<uint16_t> age;                                  // 4
  uint16_t qtype;                                             // 2
  uint16_t qclass;                                            // 2
  uint16_t origID;                                            // 2
  uint16_t origFlags;                                         // 2
  int delayMsec;
  bool ednsAdded{false};
  bool skipCache{false};
};

struct Rings {
  Rings()
  {
    queryRing.set_capacity(10000);
    respRing.set_capacity(10000);
    pthread_rwlock_init(&queryLock, 0);
  }
  struct Query
  {
    struct timespec when;
    ComboAddress requestor;
    DNSName name;
    uint16_t size;
    uint16_t qtype;
    struct dnsheader dh;
  };
  boost::circular_buffer<Query> queryRing;
  struct Response
  {
    struct timespec when;
    ComboAddress requestor;
    DNSName name;
    uint16_t qtype;
    unsigned int usec;
    unsigned int size;
    struct dnsheader dh;
    ComboAddress ds; // who handled it
  };
  boost::circular_buffer<Response> respRing;
  std::mutex respMutex;
  pthread_rwlock_t queryLock;

  vector<pair<unsigned int, ComboAddress> > getTopBandwidth(unsigned int numentries);
  unsigned int numDistinctRequestors();
};

extern Rings g_rings;

struct ClientState
{
  ComboAddress local;
#ifdef HAVE_DNSCRYPT
  DnsCryptContext* dnscryptCtx{0};
#endif
  std::atomic<uint64_t> queries{0};
  int udpFD{-1};
  int tcpFD{-1};
};

class TCPClientCollection {
  std::vector<int> d_tcpclientthreads;
  std::atomic<uint64_t> d_pos{0};
public:
  std::atomic<uint64_t> d_queued{0}, d_numthreads{0};
  uint64_t d_maxthreads{0};

  TCPClientCollection(size_t maxThreads)
  {
    d_maxthreads = maxThreads;
    d_tcpclientthreads.reserve(maxThreads);
  }

  int getThread()
  {
    uint64_t pos = d_pos++;
    ++d_queued;
    return d_tcpclientthreads[pos % d_numthreads];
  }
  void addTCPClientThread();
};

extern std::shared_ptr<TCPClientCollection> g_tcpclientthreads;

struct DownstreamState
{
  DownstreamState(const ComboAddress& remote_, const ComboAddress& sourceAddr_, unsigned int sourceItf);
  DownstreamState(const ComboAddress& remote_): DownstreamState(remote_, ComboAddress(), 0) {}
  ~DownstreamState()
  {
    if (fd >= 0)
      close(fd);
  }

  int fd{-1};
  std::thread tid;
  ComboAddress remote;
  QPSLimiter qps;
  vector<IDState> idStates;
  ComboAddress sourceAddr;
  DNSName checkName{"a.root-servers.net."};
  QType checkType{QType::A};
  std::atomic<uint64_t> idOffset{0};
  std::atomic<uint64_t> sendErrors{0};
  std::atomic<uint64_t> outstanding{0};
  std::atomic<uint64_t> reuseds{0};
  std::atomic<uint64_t> queries{0};
  struct {
    std::atomic<uint64_t> sendErrors{0};
    std::atomic<uint64_t> reuseds{0};
    std::atomic<uint64_t> queries{0};
  } prev;
  string name;
  double queryLoad{0.0};
  double dropRate{0.0};
  double latencyUsec{0.0};
  int order{1};
  int weight{1};
  int tcpRecvTimeout{30};
  int tcpSendTimeout{30};
  unsigned int sourceItf{0};
  uint16_t retries{5};
  uint8_t currentCheckFailures{0};
  uint8_t maxCheckFailures{1};
  StopWatch sw;
  set<string> pools;
  enum class Availability { Up, Down, Auto} availability{Availability::Auto};
  bool mustResolve{false};
  bool upStatus{false};
  bool useECS{false};
  bool isUp() const
  {
    if(availability == Availability::Down)
      return false;
    if(availability == Availability::Up)
      return true;
    return upStatus;
  }
  void setUp() { availability = Availability::Up; }
  void setDown() { availability = Availability::Down; }
  void setAuto() { availability = Availability::Auto; }
  string getName() const {
    if (name.empty()) {
      return remote.toStringWithPort();
    }
    return name;
  }
  string getNameWithAddr() const {
    if (name.empty()) {
      return remote.toStringWithPort();
    }
    return name + " (" + remote.toStringWithPort()+ ")";
  }

};
using servers_t =vector<std::shared_ptr<DownstreamState>>;

struct DNSQuestion
{
  DNSQuestion(const DNSName* name, uint16_t type, uint16_t class_, const ComboAddress* lc, const ComboAddress* rem, struct dnsheader* header, size_t bufferSize, uint16_t queryLen, bool isTcp): qname(name), qtype(type), qclass(class_), local(lc), remote(rem), dh(header), size(bufferSize), len(queryLen), tcp(isTcp) {};

  const DNSName* qname;
  const uint16_t qtype;
  const uint16_t qclass;
  const ComboAddress* local;
  const ComboAddress* remote;
  struct dnsheader* dh;
  size_t size;
  uint16_t len;
  const bool tcp;
  bool skipCache{false};
};

template <class T> using NumberedVector = std::vector<std::pair<unsigned int, T> >;

void* responderThread(std::shared_ptr<DownstreamState> state);
extern std::mutex g_luamutex;
extern LuaContext g_lua;
extern std::string g_outputBuffer; // locking for this is ok, as locked by g_luamutex

class DNSRule
{
public:
  virtual bool matches(const DNSQuestion* dq) const =0;
  virtual string toString() const = 0;
  mutable std::atomic<uint64_t> d_matches{0};
};

/* so what could you do: 
   drop, 
   fake up nxdomain, 
   provide actual answer, 
   allow & and stop processing, 
   continue processing, 
   modify header:    (servfail|refused|notimp), set TC=1,
   send to pool */

class DNSAction
{
public:
  enum class Action { Drop, Nxdomain, Spoof, Allow, HeaderModify, Pool, Delay, None};
  virtual Action operator()(DNSQuestion*, string* ruleresult) const =0;
  virtual string toString() const = 0;
};

using NumberedServerVector = NumberedVector<shared_ptr<DownstreamState>>;
typedef std::function<shared_ptr<DownstreamState>(const NumberedServerVector& servers, const DNSQuestion*)> policyfunc_t;

struct ServerPolicy
{
  string name;
  policyfunc_t policy;
};

struct ServerPool
{
  const std::shared_ptr<DNSDistPacketCache> getCache() const { return packetCache; };

  NumberedVector<shared_ptr<DownstreamState>> servers;
  std::shared_ptr<DNSDistPacketCache> packetCache{nullptr};
};
using pools_t=map<std::string,std::shared_ptr<ServerPool>>;
void addServerToPool(pools_t& pools, const string& poolName, std::shared_ptr<DownstreamState> server);
void removeServerFromPool(pools_t& pools, const string& poolName, std::shared_ptr<DownstreamState> server);

struct CarbonConfig
{
  vector<ComboAddress> servers;
  std::string ourname;
  unsigned int interval{30};
};

enum ednsHeaderFlags {
  EDNS_HEADER_FLAG_NONE = 0,
  EDNS_HEADER_FLAG_DO = 32768
};

extern GlobalStateHolder<CarbonConfig> g_carbon;
extern GlobalStateHolder<ServerPolicy> g_policy;
extern GlobalStateHolder<servers_t> g_dstates;
extern GlobalStateHolder<pools_t> g_pools;
extern GlobalStateHolder<vector<pair<std::shared_ptr<DNSRule>, std::shared_ptr<DNSAction> > > > g_rulactions;
extern GlobalStateHolder<NetmaskGroup> g_ACL;

extern ComboAddress g_serverControl; // not changed during runtime

extern std::vector<std::tuple<ComboAddress, bool, bool>> g_locals; // not changed at runtime (we hope XXX)
extern vector<ClientState*> g_frontends;
extern std::string g_key; // in theory needs locking
extern bool g_truncateTC;
extern bool g_fixupCase;
extern int g_tcpRecvTimeout;
extern int g_tcpSendTimeout;
extern uint16_t g_maxOutstanding;
extern std::atomic<bool> g_configurationDone;
extern uint64_t g_maxTCPClientThreads;
extern uint64_t g_maxTCPQueuedConnections;
extern std::atomic<uint16_t> g_cacheCleaningDelay;
extern uint16_t g_ECSSourcePrefixV4;
extern uint16_t g_ECSSourcePrefixV6;
extern bool g_ECSOverride;
extern bool g_verboseHealthChecks;
extern uint32_t g_staleCacheEntriesTTL;

struct dnsheader;

void controlThread(int fd, ComboAddress local);
vector<std::function<void(void)>> setupLua(bool client, const std::string& config);
std::shared_ptr<ServerPool> getPool(const pools_t& pools, const std::string& poolName);
std::shared_ptr<ServerPool> createPoolIfNotExists(pools_t& pools, const string& poolName);
const NumberedServerVector& getDownstreamCandidates(const pools_t& pools, const std::string& poolName);

std::shared_ptr<DownstreamState> firstAvailable(const NumberedServerVector& servers, const DNSQuestion* dq);

std::shared_ptr<DownstreamState> leastOutstanding(const NumberedServerVector& servers, const DNSQuestion* dq);
std::shared_ptr<DownstreamState> wrandom(const NumberedServerVector& servers, const DNSQuestion* dq);
std::shared_ptr<DownstreamState> whashed(const NumberedServerVector& servers, const DNSQuestion* dq);
std::shared_ptr<DownstreamState> roundrobin(const NumberedServerVector& servers, const DNSQuestion* dq);
int getEDNSZ(const char* packet, unsigned int len);
void spoofResponseFromString(DNSQuestion& dq, const string& spoofContent);
uint16_t getEDNSOptionCode(const char * packet, size_t len);
void dnsdistWebserverThread(int sock, const ComboAddress& local, const string& password, const string& apiKey);
bool getMsgLen32(int fd, uint32_t* len);
bool putMsgLen32(int fd, uint32_t len);
void* tcpAcceptorThread(void* p);

void moreLua(bool client);
void doClient(ComboAddress server, const std::string& command);
void doConsole();
void controlClientThread(int fd, ComboAddress client);
extern "C" {
char** my_completion( const char * text , int start,  int end);
}
void setLuaNoSideEffect(); // if nothing has been declared, set that there are no side effects
void setLuaSideEffect();   // set to report a side effect, cancelling all _no_ side effect calls
bool getLuaNoSideEffect(); // set if there were only explicit declarations of _no_ side effect
void resetLuaSideEffect(); // reset to indeterminate state

#ifdef HAVE_DNSCRYPT
extern std::vector<std::tuple<ComboAddress,DnsCryptContext,bool>> g_dnsCryptLocals;

int handleDnsCryptQuery(DnsCryptContext* ctx, char* packet, uint16_t len, std::shared_ptr<DnsCryptQuery>& query, uint16_t* decryptedQueryLen, bool tcp, std::vector<uint8_t>& reponse);
#endif
Commit	Line	Data
df111b53	1	#pragma once
11e1e08b	2	#include "config.h"
df111b53	3	#include "ext/luawrapper/include/LuaContext.hpp"
	4	#include <time.h>
	5	#include "misc.hh"
	6	#include "iputils.hh"
	7	#include "dnsname.hh"
	8	#include <atomic>
	9	#include <boost/circular_buffer.hpp>
e16fd59c	10	#include <boost/variant.hpp>
df111b53	11	#include <mutex>
df111b53	12	#include <thread>
ecbe9133	13	#include "sholder.hh"
11e1e08b	14	#include "dnscrypt.hh"
886e2cf2	15	#include "dnsdist-cache.hh"
1ea747c0	16
42fae326	17	void* carbonDumpThread();
61d1b966	18	uint64_t uptimeOfProcess(const std::string& str);
bd1c631b	19
78ffa782	20	struct DynBlock
	21	{
	22	DynBlock& operator=(const DynBlock& rhs)
	23	{
	24	reason=rhs.reason;
	25	until=rhs.until;
	26	blocks.store(rhs.blocks);
	27	return *this;
	28	}
	29
	30	string reason;
	31	struct timespec until;
	32	mutable std::atomic<unsigned int> blocks;
	33	};
	34
	35	extern GlobalStateHolder<NetmaskTree<DynBlock>> g_dynblockNMG;
f758857a	36
	37	extern vector<pair<struct timeval, std::string> > g_confDelta;
	38
e48090d1	39	struct DNSDistStats
e48090d1	40	{
6ad8b29a	41	using stat_t=std::atomic<uint64_t>; // aww yiss ;-)
e48090d1	42	stat_t responses{0};
	43	stat_t servfailResponses{0};
	44	stat_t queries{0};
e73ec7d3	45	stat_t nonCompliantQueries{0};
d08b1cdf	46	stat_t nonCompliantResponses{0};
643a182a	47	stat_t rdQueries{0};
2efd427d	48	stat_t emptyQueries{0};
e48090d1	49	stat_t aclDrops{0};
e48090d1	50	stat_t blockFilter{0};
bd1c631b	51	stat_t dynBlocked{0};
e48090d1	52	stat_t ruleDrop{0};
	53	stat_t ruleNXDomain{0};
	54	stat_t selfAnswered{0};
	55	stat_t downstreamTimeouts{0};
	56	stat_t downstreamSendErrors{0};
6ad8b29a	57	stat_t truncFail{0};
b8bc7e61	58	stat_t noPolicy{0};
886e2cf2 RG	59	stat_t cacheHits{0};
886e2cf2 RG	60	stat_t cacheMisses{0};
42fae326	61	stat_t latency0_1{0}, latency1_10{0}, latency10_50{0}, latency50_100{0}, latency100_1000{0}, latencySlow{0};
e48090d1	62
e16fd59c	63	double latencyAvg100{0}, latencyAvg1000{0}, latencyAvg10000{0}, latencyAvg1000000{0};
a1a787dc	64	typedef std::function<uint64_t(const std::string&)> statfunction_t;
a1a787dc	65	typedef boost::variant<stat_t, double, statfunction_t> entry_t;
e16fd59c	66	std::vector<std::pair<std::string, entry_t>> entries{
42fae326	67	{"responses", &responses}, {"servfail-responses", &servfailResponses},
	68	{"queries", &queries}, {"acl-drops", &aclDrops},
	69	{"block-filter", &blockFilter}, {"rule-drop", &ruleDrop},
	70	{"rule-nxdomain", &ruleNXDomain}, {"self-answered", &selfAnswered},
	71	{"downstream-timeouts", &downstreamTimeouts}, {"downstream-send-errors", &downstreamSendErrors},
	72	{"trunc-failures", &truncFail}, {"no-policy", &noPolicy},
	73	{"latency0-1", &latency0_1}, {"latency1-10", &latency1_10},
	74	{"latency10-50", &latency10_50}, {"latency50-100", &latency50_100},
e16fd59c	75	{"latency100-1000", &latency100_1000}, {"latency-slow", &latencySlow},
e16fd59c	76	{"latency-avg100", &latencyAvg100}, {"latency-avg1000", &latencyAvg1000},
a1a787dc	77	{"latency-avg10000", &latencyAvg10000}, {"latency-avg1000000", &latencyAvg1000000},
61d1b966	78	{"uptime", uptimeOfProcess},
a9b6db56	79	{"real-memory-usage", getRealMemoryUsage},
a2aa00ed	80	{"noncompliant-queries", &nonCompliantQueries},
d08b1cdf	81	{"noncompliant-responses", &nonCompliantResponses},
643a182a	82	{"rdqueries", &rdQueries},
2efd427d	83	{"empty-queries", &emptyQueries},
886e2cf2 RG	84	{"cache-hits", &cacheHits},
886e2cf2 RG	85	{"cache-misses", &cacheMisses},
4f99f3d3 RG	86	{"cpu-user-msec", getCPUTimeUser},
4f99f3d3 RG	87	{"cpu-sys-msec", getCPUTimeSystem},
bd1c631b	88	{"fd-usage", getOpenFileDescriptors}, {"dyn-blocked", &dynBlocked},
bd1c631b	89	{"dyn-block-nmg-size", [](const std::string&) { return g_dynblockNMG.getLocal()->size(); }}
42fae326	90	};
e48090d1	91	};
e48090d1	92
e16fd59c	93
e48090d1	94	extern struct DNSDistStats g_stats;
e48090d1	95
638184e9	96
df111b53	97	struct StopWatch
	98	{
	99	#ifndef CLOCK_MONOTONIC_RAW
	100	#define CLOCK_MONOTONIC_RAW CLOCK_MONOTONIC
	101	#endif
	102	struct timespec d_start{0,0};
	103	void start() {
	104	if(clock_gettime(CLOCK_MONOTONIC_RAW, &d_start) < 0)
	105	unixDie("Getting timestamp");
	106
	107	}
	108
	109	double udiff() const {
	110	struct timespec now;
	111	if(clock_gettime(CLOCK_MONOTONIC_RAW, &now) < 0)
	112	unixDie("Getting timestamp");
	113
	114	return 1000000.0*(now.tv_sec - d_start.tv_sec) + (now.tv_nsec - d_start.tv_nsec)/1000.0;
	115	}
	116
	117	double udiffAndSet() {
	118	struct timespec now;
	119	if(clock_gettime(CLOCK_MONOTONIC_RAW, &now) < 0)
	120	unixDie("Getting timestamp");
	121
	122	auto ret= 1000000.0*(now.tv_sec - d_start.tv_sec) + (now.tv_nsec - d_start.tv_nsec)/1000.0;
	123	d_start = now;
	124	return ret;
	125	}
	126
	127	};
	128
	129	class QPSLimiter
	130	{
	131	public:
	132	QPSLimiter()
	133	{
	134	}
	135
	136	QPSLimiter(unsigned int rate, unsigned int burst) : d_rate(rate), d_burst(burst), d_tokens(burst)
	137	{
	138	d_passthrough=false;
	139	d_prev.start();
	140	}
	141
	142	unsigned int getRate() const
	143	{
	144	return d_passthrough? 0 : d_rate;
	145	}
	146
	147	int getPassed() const
	148	{
	149	return d_passed;
	150	}
	151	int getBlocked() const
	152	{
	153	return d_blocked;
	154	}
	155
ecbe9133	156	bool check() const // this is not quite fair
df111b53	157	{
	158	if(d_passthrough)
	159	return true;
	160	auto delta = d_prev.udiffAndSet();
	161
	162	d_tokens += 1.0d_rate (delta/1000000.0);
	163
	164	if(d_tokens > d_burst)
	165	d_tokens = d_burst;
	166
	167	bool ret=false;
	168	if(d_tokens >= 1.0) { // we need this because burst=1 is weird otherwise
	169	ret=true;
	170	--d_tokens;
	171	d_passed++;
	172	}
	173	else
	174	d_blocked++;
	175
	176	return ret;
	177	}
	178	private:
	179	bool d_passthrough{true};
	180	unsigned int d_rate;
	181	unsigned int d_burst;
ecbe9133	182	mutable double d_tokens;
	183	mutable StopWatch d_prev;
	184	mutable unsigned int d_passed{0};
	185	mutable unsigned int d_blocked{0};
df111b53	186	};
df111b53	187
df111b53	188	struct IDState
df111b53	189	{
c9ba8478	190	IDState() : origFD(-1), delayMsec(0) { origDest.sin4.sin_family = 0;}
df111b53	191	IDState(const IDState& orig)
	192	{
	193	origFD = orig.origFD;
	194	origID = orig.origID;
	195	origRemote = orig.origRemote;
549d63c9	196	origDest = orig.origDest;
7b3865cd	197	delayMsec = orig.delayMsec;
df111b53	198	age.store(orig.age.load());
	199	}
	200
2bf26975	201	int origFD; // set to <0 to indicate this state is empty // 4
	202
	203	ComboAddress origRemote; // 28
549d63c9	204	ComboAddress origDest; // 28
2bf26975	205	StopWatch sentTime; // 16
2bf26975	206	DNSName qname; // 80
11e1e08b RG	207	#ifdef HAVE_DNSCRYPT
	208	std::shared_ptr<DnsCryptQuery> dnsCryptQuery{0};
	209	#endif
886e2cf2 RG	210	std::shared_ptr<DNSDistPacketCache> packetCache{nullptr};
886e2cf2 RG	211	uint32_t cacheKey; // 8
2bf26975	212	std::atomic<uint16_t> age; // 4
2bf26975	213	uint16_t qtype; // 2
886e2cf2	214	uint16_t qclass; // 2
2bf26975	215	uint16_t origID; // 2
aeb36780	216	uint16_t origFlags; // 2
7b3865cd	217	int delayMsec;
ca404e94	218	bool ednsAdded{false};
886e2cf2	219	bool skipCache{false};
df111b53	220	};
	221
	222	struct Rings {
	223	Rings()
	224	{
df111b53	225	queryRing.set_capacity(10000);
df111b53	226	respRing.set_capacity(10000);
0e41337b	227	pthread_rwlock_init(&queryLock, 0);
df111b53	228	}
0ba5eecf	229	struct Query
	230	{
	231	struct timespec when;
	232	ComboAddress requestor;
	233	DNSName name;
03ebf8b2	234	uint16_t size;
0ba5eecf	235	uint16_t qtype;
3fcaeeac	236	struct dnsheader dh;
0ba5eecf	237	};
0ba5eecf	238	boost::circular_buffer<Query> queryRing;
df111b53	239	struct Response
df111b53	240	{
80a216c9	241	struct timespec when;
80a216c9	242	ComboAddress requestor;
df111b53	243	DNSName name;
df111b53	244	uint16_t qtype;
df111b53	245	unsigned int usec;
80a216c9	246	unsigned int size;
3fcaeeac	247	struct dnsheader dh;
2d11d1b2	248	ComboAddress ds; // who handled it
df111b53	249	};
	250	boost::circular_buffer<Response> respRing;
	251	std::mutex respMutex;
0e41337b	252	pthread_rwlock_t queryLock;
03ebf8b2	253
ddafcf1e	254	vector<pair<unsigned int, ComboAddress> > getTopBandwidth(unsigned int numentries);
03ebf8b2	255	unsigned int numDistinctRequestors();
df111b53	256	};
df111b53	257
0e41337b	258	extern Rings g_rings;
df111b53	259
8a5d5053	260	struct ClientState
	261	{
	262	ComboAddress local;
11e1e08b RG	263	#ifdef HAVE_DNSCRYPT
	264	DnsCryptContext* dnscryptCtx{0};
	265	#endif
963bef8d	266	std::atomic<uint64_t> queries{0};
a36ce055 RG	267	int udpFD{-1};
a36ce055 RG	268	int tcpFD{-1};
8a5d5053	269	};
	270
	271	class TCPClientCollection {
	272	std::vector<int> d_tcpclientthreads;
a9bf3ec4	273	std::atomic<uint64_t> d_pos{0};
8a5d5053	274	public:
a9bf3ec4	275	std::atomic<uint64_t> d_queued{0}, d_numthreads{0};
6c1ca990	276	uint64_t d_maxthreads{0};
8a5d5053	277
a9bf3ec4	278	TCPClientCollection(size_t maxThreads)
8a5d5053	279	{
6c1ca990	280	d_maxthreads = maxThreads;
a9bf3ec4	281	d_tcpclientthreads.reserve(maxThreads);
8a5d5053	282	}
8a5d5053	283
a9bf3ec4	284	int getThread()
8a5d5053	285	{
6c1ca990	286	uint64_t pos = d_pos++;
8a5d5053	287	++d_queued;
	288	return d_tcpclientthreads[pos % d_numthreads];
	289	}
	290	void addTCPClientThread();
	291	};
	292
a9bf3ec4	293	extern std::shared_ptr<TCPClientCollection> g_tcpclientthreads;
8a5d5053	294
df111b53	295	struct DownstreamState
df111b53	296	{
fbe2a2e0 RG	297	DownstreamState(const ComboAddress& remote_, const ComboAddress& sourceAddr_, unsigned int sourceItf);
fbe2a2e0 RG	298	DownstreamState(const ComboAddress& remote_): DownstreamState(remote_, ComboAddress(), 0) {}
6a62c0e3 RG	299	~DownstreamState()
	300	{
	301	if (fd >= 0)
	302	close(fd);
	303	}
df111b53	304
f99e3aaf	305	int fd{-1};
df111b53	306	std::thread tid;
	307	ComboAddress remote;
	308	QPSLimiter qps;
	309	vector<IDState> idStates;
fbe2a2e0 RG	310	ComboAddress sourceAddr;
	311	DNSName checkName{"a.root-servers.net."};
	312	QType checkType{QType::A};
df111b53	313	std::atomic<uint64_t> idOffset{0};
	314	std::atomic<uint64_t> sendErrors{0};
	315	std::atomic<uint64_t> outstanding{0};
	316	std::atomic<uint64_t> reuseds{0};
	317	std::atomic<uint64_t> queries{0};
	318	struct {
	319	std::atomic<uint64_t> sendErrors{0};
	320	std::atomic<uint64_t> reuseds{0};
	321	std::atomic<uint64_t> queries{0};
	322	} prev;
18eeccc9	323	string name;
df111b53	324	double queryLoad{0.0};
	325	double dropRate{0.0};
	326	double latencyUsec{0.0};
	327	int order{1};
	328	int weight{1};
3f6d07a4 RG	329	int tcpRecvTimeout{30};
3f6d07a4 RG	330	int tcpSendTimeout{30};
fbe2a2e0	331	unsigned int sourceItf{0};
3f6d07a4	332	uint16_t retries{5};
9e87dcb8 RG	333	uint8_t currentCheckFailures{0};
9e87dcb8 RG	334	uint8_t maxCheckFailures{1};
df111b53	335	StopWatch sw;
	336	set<string> pools;
	337	enum class Availability { Up, Down, Auto} availability{Availability::Auto};
fbe2a2e0	338	bool mustResolve{false};
df111b53	339	bool upStatus{false};
ca404e94	340	bool useECS{false};
df111b53	341	bool isUp() const
	342	{
	343	if(availability == Availability::Down)
	344	return false;
	345	if(availability == Availability::Up)
	346	return true;
	347	return upStatus;
	348	}
	349	void setUp() { availability = Availability::Up; }
	350	void setDown() { availability = Availability::Down; }
	351	void setAuto() { availability = Availability::Auto; }
18eeccc9 RG	352	string getName() const {
	353	if (name.empty()) {
	354	return remote.toStringWithPort();
	355	}
	356	return name;
	357	}
a7940c06	358	string getNameWithAddr() const {
	359	if (name.empty()) {
	360	return remote.toStringWithPort();
	361	}
	362	return name + " (" + remote.toStringWithPort()+ ")";
	363	}
	364
df111b53	365	};
df111b53	366	using servers_t =vector<std::shared_ptr<DownstreamState>>;
df111b53	367
497a6e3a RG	368	struct DNSQuestion
497a6e3a RG	369	{
cec47783	370	DNSQuestion(const DNSName* name, uint16_t type, uint16_t class_, const ComboAddress* lc, const ComboAddress* rem, struct dnsheader* header, size_t bufferSize, uint16_t queryLen, bool isTcp): qname(name), qtype(type), qclass(class_), local(lc), remote(rem), dh(header), size(bufferSize), len(queryLen), tcp(isTcp) {};
497a6e3a RG	371
	372	const DNSName* qname;
	373	const uint16_t qtype;
cec47783	374	const uint16_t qclass;
497a6e3a RG	375	const ComboAddress* local;
	376	const ComboAddress* remote;
	377	struct dnsheader* dh;
	378	size_t size;
	379	uint16_t len;
	380	const bool tcp;
886e2cf2	381	bool skipCache{false};
497a6e3a RG	382	};
497a6e3a RG	383
da4e7813	384	template <class T> using NumberedVector = std::vector<std::pair<unsigned int, T> >;
	385
	386	void* responderThread(std::shared_ptr<DownstreamState> state);
	387	extern std::mutex g_luamutex;
	388	extern LuaContext g_lua;
	389	extern std::string g_outputBuffer; // locking for this is ok, as locked by g_luamutex
	390
0940e4eb	391	class DNSRule
	392	{
	393	public:
497a6e3a	394	virtual bool matches(const DNSQuestion* dq) const =0;
0940e4eb	395	virtual string toString() const = 0;
	396	mutable std::atomic<uint64_t> d_matches{0};
	397	};
	398
	399	/* so what could you do:
	400	drop,
	401	fake up nxdomain,
	402	provide actual answer,
	403	allow & and stop processing,
	404	continue processing,
	405	modify header: (servfail\|refused\|notimp), set TC=1,
	406	send to pool */
	407
	408	class DNSAction
	409	{
	410	public:
7b3865cd	411	enum class Action { Drop, Nxdomain, Spoof, Allow, HeaderModify, Pool, Delay, None};
497a6e3a	412	virtual Action operator()(DNSQuestion, string ruleresult) const =0;
0940e4eb	413	virtual string toString() const = 0;
	414	};
	415
da4e7813	416	using NumberedServerVector = NumberedVector<shared_ptr<DownstreamState>>;
497a6e3a	417	typedef std::function<shared_ptr<DownstreamState>(const NumberedServerVector& servers, const DNSQuestion*)> policyfunc_t;
df111b53	418
	419	struct ServerPolicy
	420	{
	421	string name;
70a57b05	422	policyfunc_t policy;
df111b53	423	};
df111b53	424
886e2cf2 RG	425	struct ServerPool
	426	{
	427	const std::shared_ptr<DNSDistPacketCache> getCache() const { return packetCache; };
	428
	429	NumberedVector<shared_ptr<DownstreamState>> servers;
	430	std::shared_ptr<DNSDistPacketCache> packetCache{nullptr};
	431	};
	432	using pools_t=map<std::string,std::shared_ptr<ServerPool>>;
	433	void addServerToPool(pools_t& pools, const string& poolName, std::shared_ptr<DownstreamState> server);
	434	void removeServerFromPool(pools_t& pools, const string& poolName, std::shared_ptr<DownstreamState> server);
	435
42fae326	436	struct CarbonConfig
42fae326	437	{
3565320b	438	vector<ComboAddress> servers;
42fae326	439	std::string ourname;
	440	unsigned int interval{30};
	441	};
	442
ca404e94 RG	443	enum ednsHeaderFlags {
	444	EDNS_HEADER_FLAG_NONE = 0,
	445	EDNS_HEADER_FLAG_DO = 32768
	446	};
	447
42fae326	448	extern GlobalStateHolder<CarbonConfig> g_carbon;
ecbe9133	449	extern GlobalStateHolder<ServerPolicy> g_policy;
ecbe9133	450	extern GlobalStateHolder<servers_t> g_dstates;
886e2cf2	451	extern GlobalStateHolder<pools_t> g_pools;
0940e4eb	452	extern GlobalStateHolder<vector<pair<std::shared_ptr<DNSRule>, std::shared_ptr<DNSAction> > > > g_rulactions;
638184e9	453	extern GlobalStateHolder<NetmaskGroup> g_ACL;
2e72cc0e	454
ecbe9133	455	extern ComboAddress g_serverControl; // not changed during runtime
ecbe9133	456
4c34246d	457	extern std::vector<std::tuple<ComboAddress, bool, bool>> g_locals; // not changed at runtime (we hope XXX)
963bef8d	458	extern vector<ClientState*> g_frontends;
ecbe9133	459	extern std::string g_key; // in theory needs locking
6ad8b29a	460	extern bool g_truncateTC;
b29edbee	461	extern bool g_fixupCase;
3f6d07a4 RG	462	extern int g_tcpRecvTimeout;
3f6d07a4 RG	463	extern int g_tcpSendTimeout;
e41f8165 RG	464	extern uint16_t g_maxOutstanding;
e41f8165 RG	465	extern std::atomic<bool> g_configurationDone;
6c1ca990 RG	466	extern uint64_t g_maxTCPClientThreads;
6c1ca990 RG	467	extern uint64_t g_maxTCPQueuedConnections;
886e2cf2	468	extern std::atomic<uint16_t> g_cacheCleaningDelay;
ca404e94 RG	469	extern uint16_t g_ECSSourcePrefixV4;
	470	extern uint16_t g_ECSSourcePrefixV6;
	471	extern bool g_ECSOverride;
9e87dcb8	472	extern bool g_verboseHealthChecks;
1ea747c0	473	extern uint32_t g_staleCacheEntriesTTL;
ca404e94	474
ecbe9133	475	struct dnsheader;
	476
	477	void controlThread(int fd, ComboAddress local);
839f3021	478	vector<std::function<void(void)>> setupLua(bool client, const std::string& config);
886e2cf2 RG	479	std::shared_ptr<ServerPool> getPool(const pools_t& pools, const std::string& poolName);
	480	std::shared_ptr<ServerPool> createPoolIfNotExists(pools_t& pools, const string& poolName);
	481	const NumberedServerVector& getDownstreamCandidates(const pools_t& pools, const std::string& poolName);
da4e7813	482
497a6e3a	483	std::shared_ptr<DownstreamState> firstAvailable(const NumberedServerVector& servers, const DNSQuestion* dq);
ecbe9133	484
497a6e3a RG	485	std::shared_ptr<DownstreamState> leastOutstanding(const NumberedServerVector& servers, const DNSQuestion* dq);
	486	std::shared_ptr<DownstreamState> wrandom(const NumberedServerVector& servers, const DNSQuestion* dq);
	487	std::shared_ptr<DownstreamState> whashed(const NumberedServerVector& servers, const DNSQuestion* dq);
	488	std::shared_ptr<DownstreamState> roundrobin(const NumberedServerVector& servers, const DNSQuestion* dq);
520eb5a0	489	int getEDNSZ(const char* packet, unsigned int len);
6ca7a40a	490	void spoofResponseFromString(DNSQuestion& dq, const string& spoofContent);
ca404e94	491	uint16_t getEDNSOptionCode(const char * packet, size_t len);
87893e08	492	void dnsdistWebserverThread(int sock, const ComboAddress& local, const string& password, const string& apiKey);
6885d4bf	493	bool getMsgLen32(int fd, uint32_t* len);
6885d4bf	494	bool putMsgLen32(int fd, uint32_t len);
d8d85a30	495	void* tcpAcceptorThread(void* p);
80a216c9	496
886e2cf2	497	void moreLua(bool client);
ffb07158	498	void doClient(ComboAddress server, const std::string& command);
	499	void doConsole();
	500	void controlClientThread(int fd, ComboAddress client);
	501	extern "C" {
	502	char** my_completion( const char * text , int start, int end);
	503	}
f758857a	504	void setLuaNoSideEffect(); // if nothing has been declared, set that there are no side effects
	505	void setLuaSideEffect(); // set to report a side effect, cancelling all _no_ side effect calls
	506	bool getLuaNoSideEffect(); // set if there were only explicit declarations of _no_ side effect
	507	void resetLuaSideEffect(); // reset to indeterminate state
11e1e08b RG	508
11e1e08b RG	509	#ifdef HAVE_DNSCRYPT
4c34246d	510	extern std::vector<std::tuple<ComboAddress,DnsCryptContext,bool>> g_dnsCryptLocals;
11e1e08b RG	511
	512	int handleDnsCryptQuery(DnsCryptContext* ctx, char* packet, uint16_t len, std::shared_ptr<DnsCryptQuery>& query, uint16_t* decryptedQueryLen, bool tcp, std::vector<uint8_t>& reponse);
	513	#endif