XrdXrootdMonitor.cc

Go to the documentation of this file.

/******************************************************************************/
/*                                                                            */
/*                   X r d X r o o t d M o n i t o r . c c                    */
/*                                                                            */
/* (c) 2004 by the Board of Trustees of the Leland Stanford, Jr., University  */
/*                            All Rights Reserved                             */
/*   Produced by Andrew Hanushevsky for Stanford University under contract    */
/*              DE-AC02-76-SFO0515 with the Department of Energy              */
/*                                                                            */
/* This file is part of the XRootD software suite.                            */
/*                                                                            */
/* XRootD is free software: you can redistribute it and/or modify it under    */
/* the terms of the GNU Lesser General Public License as published by the     */
/* Free Software Foundation, either version 3 of the License, or (at your     */
/* option) any later version.                                                 */
/*                                                                            */
/* XRootD is distributed in the hope that it will be useful, but WITHOUT      */
/* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or      */
/* FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public       */
/* License for more details.                                                  */
/*                                                                            */
/* You should have received a copy of the GNU Lesser General Public License   */
/* along with XRootD in a file called COPYING.LESSER (LGPL license) and file  */
/* COPYING (GPL license).  If not, see <http://www.gnu.org/licenses/>.        */
/*                                                                            */
/* The copyright holder's institutional names and contributor's names may not */
/* be used to endorse or promote products derived from this software without  */
/* specific prior written permission of the institution or contributor.       */
/******************************************************************************/
 
#include <algorithm>
#include <cerrno>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <ctime>
#include <unistd.h>
#include <sys/types.h>
 
#include "XrdVersion.hh"
 
#include "XrdNet/XrdNetMsg.hh"
#include "XrdOuc/XrdOucEnv.hh"
#include "XrdOuc/XrdOucUtils.hh"
#include "XrdSys/XrdSysError.hh"
#include "XrdSys/XrdSysPlatform.hh"
 
#include "Xrd/XrdScheduler.hh"
#include "XrdXrootd/XrdXrootdMonitor.hh"
#include "XrdXrootd/XrdXrootdMonFile.hh"
#include "XrdXrootd/XrdXrootdTrace.hh"
 
/******************************************************************************/
/*                     S t a t i c   A l l o c a t i o n                      */
/******************************************************************************/
 
char              *XrdXrootdMonitor::idRec      = 0;
int                XrdXrootdMonitor::idLen      = 0;
char              *XrdXrootdMonitor::Dest1      = 0;
int                XrdXrootdMonitor::monMode1   = 0;
XrdNetMsg         *XrdXrootdMonitor::InetDest1  = 0;
char              *XrdXrootdMonitor::Dest2      = 0;
int                XrdXrootdMonitor::monMode2   = 0;
XrdNetMsg         *XrdXrootdMonitor::InetDest2  = 0;
XrdXrootdMonitor  *XrdXrootdMonitor::altMon     = 0;
XrdSysMutex        XrdXrootdMonitor::windowMutex;
int                XrdXrootdMonitor::monRlen    = 0;
XrdXrootdMonitor::MonRdrBuff
                   XrdXrootdMonitor::rdrMon[XrdXrootdMonitor::rdrMax];
XrdXrootdMonitor::MonRdrBuff
                  *XrdXrootdMonitor::rdrMP      = 0;
XrdSysMutex        XrdXrootdMonitor::rdrMutex;
int                XrdXrootdMonitor::monBlen    = 0;
int                XrdXrootdMonitor::lastEnt    = 0;
int                XrdXrootdMonitor::lastRnt    = 0;
int                XrdXrootdMonitor::isEnabled  = 0;
int                XrdXrootdMonitor::numMonitor = 0;
int                XrdXrootdMonitor::autoFlash  = 0;
int                XrdXrootdMonitor::autoFlush  = 600;
int                XrdXrootdMonitor::FlushTime  = 0;
int                XrdXrootdMonitor::monIdent   = 3600;
kXR_int32          XrdXrootdMonitor::currWindow = 0;
int                XrdXrootdMonitor::rdrTOD     = 0;
int                XrdXrootdMonitor::rdrWin     = 0;
int                XrdXrootdMonitor::rdrNum     = 3;
kXR_int32          XrdXrootdMonitor::sizeWindow = 60;
char               XrdXrootdMonitor::monINFO    = 0;
char               XrdXrootdMonitor::monIO      = 0;
char               XrdXrootdMonitor::monFILE    = 0;
char               XrdXrootdMonitor::monREDR    = 0;
char               XrdXrootdMonitor::monUSER    = 0;
char               XrdXrootdMonitor::monAUTH    = 0;
char               XrdXrootdMonitor::monACTIVE  = 0;
char               XrdXrootdMonitor::monFSTAT   = 0;
char               XrdXrootdMonitor::monCLOCK   = 0;
 
/******************************************************************************/
/*                               G l o b a l s                                */
/******************************************************************************/
  
extern          XrdSysTrace        XrdXrootdTrace;
 
namespace XrdXrootdMonInfo
{
 
inline static int32_t  InitStartTime()
{
  return htonl( time( 0 ) );
}
 
XrdScheduler   *Sched     = 0;
XrdSysError    *eDest     = 0;
char           *monHost   = 0;
char           *kySID     = 0;
long long       mySID     = 0;
int32_t         startTime = InitStartTime();
int             kySIDSZ   = 0;
XrdSysMutex     seqMutex;
 
char           *SidCGI[4] = {0};
int             LidCGI[4] = {0};
char           *SidJSON[4]= {0}; // 0:sidsite 1:sidhostid 2:sidinst 3:sidfull
int             LidJSON[4]= {0};
}
 
using namespace XrdXrootdMonInfo;
 
/******************************************************************************/
/*                         L o c a l   D e f i n e s                          */
/******************************************************************************/
 
#define setTMark(TM_mb, TM_en, TM_tm) \
           TM_mb->info[TM_en].arg0.val    = mySID; \
           TM_mb->info[TM_en].arg0.id[0]  = XROOTD_MON_WINDOW; \
           TM_mb->info[TM_en].arg1.Window = \
           TM_mb->info[TM_en].arg2.Window = static_cast<kXR_int32>(ntohl(TM_tm));
 
#define setTMurk(TM_mb, TM_en, TM_tm) \
           TM_mb->info[TM_en].arg0.Window = rdrWin; \
           TM_mb->info[TM_en].arg1.Window = static_cast<kXR_int32>(TM_tm);
  
/******************************************************************************/
/*                         L o c a l   C l a s s e s                          */
/******************************************************************************/
/******************************************************************************/
/*                X r d X r o o t d M o n i t o r _ I d e n t                 */
/******************************************************************************/
  
class XrdXrootdMonitor_Ident : public XrdJob
{
public:
 
void          DoIt() {
                      if (idInt >= 0)
                         {if (doIdnt) XrdXrootdMonitor::Ident();
                          if (doHail) doHail = XrdXrootdMonitor::Hello::Hail();
                         }
                      if ((doIdnt || doHail) && idInt > 0)
                         Sched->Schedule((XrdJob *)this, time(0)+idInt);
                     }
 
      XrdXrootdMonitor_Ident(int idt, bool ison) : XrdJob("monitor ident"),
                             idInt(idt), doIdnt(ison), doHail(true) {}
     ~XrdXrootdMonitor_Ident() {}
 
private:
int           idInt;
bool          doIdnt;
bool          doHail;
};
 
/******************************************************************************/
/*           C l a s s   X r d X r o o t d M o n i t o r _ T i c k            */
/******************************************************************************/
 
class XrdXrootdMonitor_Tick : public XrdJob
{
public:
 
void          DoIt() {
#ifndef NODEBUG
                      const char *TraceID = "MonTick";
#endif
                      time_t Now = XrdXrootdMonitor::Tick();
                      if (Window && Now)
                         Sched->Schedule((XrdJob *)this, Now+Window);
                         else {TRACE(DEBUG, "Monitor clock stopping.");}
                     }
 
void          Set(XrdScheduler *sp, int intvl) {Sched = sp; Window = intvl;}
 
      XrdXrootdMonitor_Tick() : XrdJob("monitor window clock"),
                                Sched(0), Window(0) {}
     ~XrdXrootdMonitor_Tick() {}
 
private:
XrdScheduler  *Sched;     // System scheduler
int            Window;
};
 
/******************************************************************************/
/*            C l a s s   X r d X r o o t d M o n i t o r L o c k             */
/******************************************************************************/
  
class XrdXrootdMonitorLock
{
public:
 
static void Lock()   {monLock.Lock();}
 
static void UnLock() {monLock.UnLock();}
 
       XrdXrootdMonitorLock(XrdXrootdMonitor *theMonitor)
                {if (theMonitor != XrdXrootdMonitor::altMon) unLock = 0;
                    else {unLock = 1; monLock.Lock();}
                }
      ~XrdXrootdMonitorLock() {if (unLock) monLock.UnLock();}
 
private:
 
static XrdSysMutex monLock;
       char        unLock;
};
 
XrdSysMutex XrdXrootdMonitorLock::monLock;
 
/******************************************************************************/
/*               X r d X r o o t d M o n i t o r : : H e l l o                */
/******************************************************************************/
  
XrdXrootdMonitor::Hello::Hello(const char *dest, char mode)
                              : Next(0), theDest(0), theMode(0)
{
   if (dest)
      {Hello *nP = First;
       while(nP) {if (!strcmp(dest, nP->theDest) && mode == theMode) return;
                  nP = nP->Next;
                 }
       Next    = First;
       First   = this;
       theDest = strdup(dest);
       theMode = mode;
      }
}
 
/******************************************************************************/
/*         X r d X r o o t d M o n i t o r : : H e l l o : : H a i l          */
/******************************************************************************/
 
XrdXrootdMonitor::Hello *XrdXrootdMonitor::Hello::First = 0;
  
bool XrdXrootdMonitor::Hello::Hail()
{
   Hello *nP = First;
 
// Call all the registered ident methods
//
   while(nP) {nP->Ident(); nP = nP->Next;}
 
// Indicate whether or not anything would have been sent
//
   return First != 0;
}
  
/******************************************************************************/
/*       X r d X r o o t d M o n i t o r : : U s e r : : D i s a b l e        */
/******************************************************************************/
 
void XrdXrootdMonitor::User::Disable()
{
   if (Agent)
      {XrdXrootdMonitor::unAlloc(Agent); Agent = 0;}
   Fops = Iops = 0;
}
  
/******************************************************************************/
/*        X r d X r o o t d M o n i t o r : : U s e r : : E n a b l e         */
/******************************************************************************/
 
void XrdXrootdMonitor::User::Enable()
{
   if (Agent || (Agent = XrdXrootdMonitor::Alloc(1)))
      {Iops = XrdXrootdMonitor::monIO;
       Fops = XrdXrootdMonitor::monFILE;
      } else Iops = Fops = 0;
}
  
/******************************************************************************/
/*      X r d X r o o t d M o n i t o r : : U s e r : : R e g i s t e r       */
/******************************************************************************/
 
void XrdXrootdMonitor::User::Register(const char *Uname, 
                                      const char *Hname,
                                      const char *Pname, unsigned int xSID)
{
#ifndef NODEBUG
    const char *TraceID = "Monitor";
#endif
   char *dotP, *colonP, *atP;
   char  uBuff[1024], tBuff[1024], sBuff[64];
 
// Decode the user name as a.b:c@d and remap it for monitoring as
// <protocol>/a.{b|xSID}:<kySID>@host
//
   snprintf(tBuff, sizeof(tBuff), "%s", Uname);
   if ((dotP = index(tBuff, '.')) && (colonP = index(dotP+1, ':')) && 
       (atP = index(colonP+1, '@')))
      {*dotP = 0; *colonP = 0; *atP = 0;
       if (xSID)  
          {snprintf(sBuff, sizeof(sBuff), " %u", xSID);
           dotP = sBuff;
          }
 
       int n = snprintf(uBuff, sizeof(uBuff), "%s/%s.%s:%s@%s", Pname, tBuff,
                        dotP+1, kySID, atP+1);
 
       if (n < 0 || n >= (int) sizeof(uBuff))
         TRACE(LOGIN, "Login ID was truncated: " << uBuff);
 
       if (xSID) {TRACE(LOGIN,"Register remap "<<Uname<<" -> "<<uBuff);}
      } else snprintf(uBuff, sizeof(uBuff), "%s/%s", Pname, Uname);
 
// Generate a monitor identity for this user. We do not assign a dictioary
// identifier unless this entry is reported.
//
   Agent = XrdXrootdMonitor::Alloc();
   Did   = 0;
   Len   = strlen(uBuff);
   Name  = strdup(uBuff);
   Iops  = XrdXrootdMonitor::monIO;
   Fops  = XrdXrootdMonitor::monFILE;
}
 
/******************************************************************************/
/*                                R e p o r t                                 */
/******************************************************************************/
  
void XrdXrootdMonitor::User::Report(int eCode, int aCode)
{
   char buff[1024];
 
   snprintf(buff, sizeof(buff), "&Uc=%u&Ec=%d&Ac=%d", ntohl(Did), eCode, aCode);
 
   XrdXrootdMonitor::Map(XROOTD_MON_MAPUEAC,*this,buff);
}
  
/******************************************************************************/
  
bool XrdXrootdMonitor::User::Report(WhatInfo infoT, const char *info)
{
   char buff[4096];
 
// Currently we support only the token external report
//
   if (infoT != TokenInfo) return false;
 
   snprintf(buff, sizeof(buff), "&Uc=%u%s%s", ntohl(Did),
                                (*info == '&' ? "" : "&"), info);
 
   XrdXrootdMonitor::Map(XROOTD_MON_MAPTOKN,*this,buff);
 
   return true;
}
/******************************************************************************/
/*                           C o n s t r u c t o r                            */
/******************************************************************************/
  
XrdXrootdMonitor::XrdXrootdMonitor()
{
   kXR_int32 localWindow;
 
// Initialize last window to force a mark as well as the local window
//
   lastWindow  = 0;
   localWindow = currWindow;
 
// Allocate a monitor buffer
//
   if (posix_memalign((void **)&monBuff, getpagesize(), monBlen))
      eDest->Emsg("Monitor", "Unable to allocate monitor buffer.");
      else {nextEnt = 1;
            setTMark(monBuff, 0, localWindow);
           }
}
 
/******************************************************************************/
/*                            D e s t r u c t o r                             */
/******************************************************************************/
 
XrdXrootdMonitor::~XrdXrootdMonitor()
{
// Release buffer
   if (monBuff) {Flush(); free(monBuff);}
}
 
/******************************************************************************/
/*                                 a p p I D                                  */
/******************************************************************************/
  
void XrdXrootdMonitor::appID(char *id)
{
   static const int apInfoSize = sizeof(XrdXrootdMonTrace)-4;
 
// Application ID's are only meaningful for io event recording
//
   if (this == altMon || !*id) return;
 
// Fill out the monitor record
//
   if (lastWindow != currWindow) Mark();
      else if (nextEnt == lastEnt) Flush();
   monBuff->info[nextEnt].arg0.id[0]  = XROOTD_MON_APPID;
   strncpy((char *)(&(monBuff->info[nextEnt])+4), id, apInfoSize);
}
 
/******************************************************************************/
/*                                 A l l o c                                  */
/******************************************************************************/
  
XrdXrootdMonitor *XrdXrootdMonitor::Alloc(int force)
{
   XrdXrootdMonitor *mp;
   int lastVal;
 
// If enabled, create a new object (if possible). If we are not monitoring
// i/o then return the global object.
//
// if (!isEnabled || (isEnabled < 0 && !force)) mp = 0;
   if (!isEnabled) mp = 0;
      else if (!monIO) mp = altMon;
              else if ((mp = new XrdXrootdMonitor()))
                      if (!(mp->monBuff)) {delete mp; mp = 0;}
 
// Check if we should turn on the monitor clock
//
   if (mp && isEnabled < 0)
      {windowMutex.Lock();
       lastVal = numMonitor; numMonitor++;
       if (!lastVal && !monREDR) startClock();
       windowMutex.UnLock();
      }
 
// All done
//
   return mp;
}
 
/******************************************************************************/
/*                                 C l o s e                                  */
/******************************************************************************/
 
void XrdXrootdMonitor::Close(kXR_unt32 dictid, long long rTot, long long wTot)
{
  XrdXrootdMonitorLock mLock(this);
  unsigned int rVal, wVal;
 
// Fill out the monitor record (we allow the compiler to correctly cast data)
//
   if (lastWindow != currWindow) Mark();
      else if (nextEnt == lastEnt) Flush();
   monBuff->info[nextEnt].arg0.id[0]    = XROOTD_MON_CLOSE;
   monBuff->info[nextEnt].arg0.id[1]    = do_Shift(rTot, rVal);
   monBuff->info[nextEnt].arg0.rTot[1]  = htonl(rVal);
   monBuff->info[nextEnt].arg0.id[2]    = do_Shift(wTot, wVal);
   monBuff->info[nextEnt].arg0.id[3]    = 0;
   monBuff->info[nextEnt].arg1.wTot     = htonl(wVal);
   monBuff->info[nextEnt++].arg2.dictid = dictid;
 
// Check if we need to duplicate this entry
//
   if (altMon && this != altMon) altMon->Dup(&monBuff->info[nextEnt-1]);
}
 
/******************************************************************************/
/*                              C o p y T e x t                               */
/******************************************************************************/
 
int XrdXrootdMonitor::CopyText(char *buff, int room, const char *text, int tLen)
{
// A record that binds text to a dictid must fit its buffer. So truncate the
// text, rather than drop the record, and null pad the rest of the buffer. At
// least one null byte always ends the text. The caller gives the length of
// the text, because it needs the length to size the record.
//
   if (room <= 0) return 0;
   if (tLen > room - 1) tLen = room - 1;
   memcpy(buff, text, tLen);
   memset(buff + tLen, 0, room - tLen);
   return tLen;
}
 
/******************************************************************************/
/*                              D e f a u l t s                               */
/******************************************************************************/
 
// This version must be called after the subsequent version!
 
void XrdXrootdMonitor::Defaults(char *dest1, int mode1, char *dest2, int mode2)
{
   int  mmode;
 
// If there are no destination then only g-stream events may be enabled.
// Otherwise, sort out the destinations relative to modes.
//
   if (!dest1 && !dest2) {isEnabled = 0; return;}
   if (!dest1) {dest1 = dest2; dest2 = 0; mode1 |= mode2; mode2 = 0;}
 
// Set the default destinations (caller supplied strdup'd strings)
//
   if (Dest1) free(Dest1);
   Dest1 = dest1; monMode1 = mode1;
   if (Dest2) free(Dest2);
   Dest2 = dest2; monMode2 = mode2;
 
// Set overall monitor mode
//
   mmode     = mode1 | mode2;
   monACTIVE = (mmode                   ? 1 : 0);
   isEnabled = (mmode & XROOTD_MON_ALL  ? 1 :-1);
   monIO     = (mmode & XROOTD_MON_IO   ? 1 : 0);
   monIO     = (mmode & XROOTD_MON_IOV  ? 2 : monIO);
   monINFO   = (mmode & XROOTD_MON_INFO ? 1 : 0);
   monFILE   = (mmode & XROOTD_MON_FILE ? 1 : 0) | monIO;
   monREDR   = (mmode & XROOTD_MON_REDR ? 1 : 0);
   monUSER   = (mmode & XROOTD_MON_USER ? 1 : 0);
   monAUTH   = (mmode & XROOTD_MON_AUTH ? 1 : 0);
   monFSTAT  = (mmode & XROOTD_MON_FSTA && monFSTAT ? 1 : 0);
 
// Compute whether or not we need the clock running
//
   if (monREDR || (isEnabled > 0 && (monIO || monFILE))) monCLOCK = 1;
 
// Check where user information should go
//
   if (((mode1 & XROOTD_MON_IO) && (mode1 & XROOTD_MON_USER))
   ||  ((mode2 & XROOTD_MON_IO) && (mode2 & XROOTD_MON_USER)))
      {if ((!(mode1 & XROOTD_MON_IO) && (mode1 & XROOTD_MON_USER))
       ||  (!(mode2 & XROOTD_MON_IO) && (mode2 & XROOTD_MON_USER))) monUSER = 3;
          else monUSER = 2;
      }
 
// If we are monitoring redirections then set an envar saying how often idents
// should be sent (this also tips off other layers to handle such monitoring)
//
   if (monREDR) XrdOucEnv::Export("XRDMONRDR", monIdent);
}
 
/******************************************************************************/
 
void XrdXrootdMonitor::Defaults(int msz,   int rsz,   int wsz,
                                int flush, int flash, int idt, int rnm,
                                int fbsz, int fsint, int fsopt, int fsion)
{
 
// Set default window size and flush time
//
   sizeWindow = (wsz <= 0 ? 60 : wsz);
   autoFlush  = (flush <= 0 ? 600 : flush);
   autoFlash  = (flash <= 0 ?   0 : flash);
   monIdent   =  idt;
   rdrNum     = (rnm   <= 0 || rnm > rdrMax ? 3 : rnm);
   rdrWin     = (sizeWindow > 16777215 ? 16777215 : sizeWindow);
   rdrWin     = htonl(rdrWin);
 
// Set the fstat defaults
//
   XrdXrootdMonFile::Defaults(fsint, fsopt, fsion, fbsz);
   monFSTAT = fsint != 0;
 
// Set default monitor buffer size. The configuration accepts up to 65535
// bytes, but the buffer goes out as one datagram, so limit it to maxDgram.
//
   if (msz <= 0) msz = 16384;
      else if (msz < 1024) msz = 1024;
              else {if (msz > maxDgram) msz = maxDgram;
                    msz = msz/sizeof(XrdXrootdMonTrace)*sizeof(XrdXrootdMonTrace);
                   }
   lastEnt = (msz-sizeof(XrdXrootdMonHeader))/sizeof(XrdXrootdMonTrace);
   monBlen = (lastEnt*sizeof(XrdXrootdMonTrace))+sizeof(XrdXrootdMonHeader);
   lastEnt--;
 
// Set default monitor redirect buffer size
//
   if (rsz <= 0) rsz = 32768;
      else if (rsz < 2048) rsz = 2048;
   lastRnt = (rsz-(sizeof(XrdXrootdMonHeader) + 16))/sizeof(XrdXrootdMonRedir);
   monRlen =  (lastRnt*sizeof(XrdXrootdMonRedir))+sizeof(XrdXrootdMonHeader)+16;
   lastRnt--;
}
 
/******************************************************************************/
/*                                  D i s c                                   */
/******************************************************************************/
 
void XrdXrootdMonitor::Disc(kXR_unt32 dictid, int csec, char Flags)
{
  XrdXrootdMonitorLock mLock(this);
 
// Check if this should not be included in the io trace
//
   if (this != altMon && monUSER == 1 && altMon)
      {altMon->Disc(dictid, csec); return;}
 
// Fill out the monitor record (let compiler cast the data correctly)
//
   if (lastWindow != currWindow) Mark();
      else if (nextEnt == lastEnt) Flush();
   monBuff->info[nextEnt].arg0.rTot[0]  = 0;
   monBuff->info[nextEnt].arg0.id[0]    = XROOTD_MON_DISC;
   monBuff->info[nextEnt].arg0.id[1]    = Flags;
   monBuff->info[nextEnt].arg1.wTot     = htonl(csec);
   monBuff->info[nextEnt++].arg2.dictid = dictid;
 
// Check if we need to duplicate this entry
//
   if (altMon && this != altMon && monUSER == 3)
      altMon->Dup(&monBuff->info[nextEnt-1]);
}
  
/******************************************************************************/
/*                                   D u p                                    */
/******************************************************************************/
  
void XrdXrootdMonitor::Dup(XrdXrootdMonTrace *mrec)
{
  XrdXrootdMonitorLock mLock(this);
 
// Fill out the monitor record
//
   if (lastWindow != currWindow) Mark();
      else if (nextEnt == lastEnt) Flush();
   memcpy(&monBuff->info[nextEnt],(const void *)mrec,sizeof(XrdXrootdMonTrace));
   nextEnt++;
}
 
/******************************************************************************/
/* Private:                        F e t c h                                  */
/******************************************************************************/
  
XrdXrootdMonitor::MonRdrBuff *XrdXrootdMonitor::Fetch()
{
   MonRdrBuff *bP;
 
// Get the next available stream and promote another one
//
   rdrMutex.Lock();
   if ((bP = rdrMP)) rdrMP = rdrMP->Next;
   rdrMutex.UnLock();
   return bP;
}
 
/******************************************************************************/
/*                                  I n i t                                   */
/******************************************************************************/
  
void XrdXrootdMonitor::Init(XrdScheduler *sp,    XrdSysError *errp,
                            const char   *iHost, const char  *iProg,
                            const char   *iName, int Port)
{
   const char *cgID0 = "&site=%s";
   const char *cgID1 = "&host=%s";
   const char *cgID2 = "&port=%d&inst=%s";
   const char *cgID3 = "&pgm=%s&ver=%s";
 
   const char *jsID0 = "\"src\":{\"site\":\"%s\"}";
   const char *jsID1 = "%s\"host\":\"%s\"}";
   const char *jsID2 = "%s\"port\":%d,\"inst\":\"%s\"}";
   const char *jsID3 = "%s\"pgm\":\"%s\",\"ver\":\"%s\"}";
 
   XrdXrootdMonMap *mP;
   char       iBuff[1024], iMuff[2048], iPuff[1024];
   int        n, i, j;
 
// Set static variables
//
   Sched = sp;
   eDest = errp;
 
// Generate our server ID (the version is not part of he fingerprint)
//
   strcpy(iBuff, "=/");
   kySID = XrdOucUtils::Ident(mySID, iBuff+2, sizeof(iBuff)-2,
                              iHost, iProg, iName, Port);
   n = strlen(iBuff);
   snprintf(iBuff+n, sizeof(iBuff)-n, "&ver=%s", XrdVERSION);
 
   kySIDSZ = strlen(kySID);
   monHost = strdup(iHost);
 
// Ignore array bounds warning from gcc 12 triggered because the allocated
// memory for the XrdXrootdMonMap is smaller than sizeof(XrdXrootdMonMap)
#if defined(__GNUC__) && __GNUC__ >= 12
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Warray-bounds"
#endif
// Create identification record
//
   idLen = strlen(iBuff) + sizeof(XrdXrootdMonHeader) + sizeof(kXR_int32);
   idRec = (char *)malloc(idLen+1);
   mP = (XrdXrootdMonMap *)idRec;
   fillHeader(&(mP->hdr), XROOTD_MON_MAPIDNT, idLen);
   mP->hdr.pseq = 0;
   mP->dictid   = 0;
   strcpy(mP->info, iBuff);
#if defined(__GNUC__) && __GNUC__ >= 12
#pragma GCC diagnostic pop
#endif
 
// Generate a CGI version of all the variations
//
   const char *Site (getenv("XRDSITE") ? getenv("XRDSITE") : "");
   i = snprintf(iPuff, sizeof(iPuff), cgID0, Site);
   SidCGI[0] = strdup(iPuff);
   LidCGI[0] = strlen(iPuff);
 
   n = sizeof(iPuff)-i; j = i;
   i = snprintf(iPuff+j, n, cgID1, iHost);
   SidCGI[1] = strdup(iPuff);
   LidCGI[1] = strlen(iPuff);
 
   n -= i; j += i;
   i = snprintf(iPuff+j, n, cgID2, Port, iName);
   SidCGI[2] = strdup(iPuff);
   LidCGI[2] = strlen(iPuff);
 
   n -= i; j += i;
       snprintf(iPuff+j, n, cgID3, iProg, XrdVERSION);
   SidCGI[3] = strdup(iPuff);
   LidCGI[3] = strlen(iPuff);
 
// Generate a JSON version of all the variations.
//
   n = snprintf(iPuff, sizeof(iPuff), jsID0, Site);
   SidJSON[0] = strdup(iPuff);
   LidJSON[0] = strlen(iPuff);
 
   strcpy(iPuff+n-1, ",");
   n = snprintf(iMuff, sizeof(iMuff), jsID1, iPuff, iHost);
   SidJSON[1] = strdup(iMuff);
   LidJSON[1] = strlen(iMuff);
 
   strcpy(iMuff+n-1, ",");
   n = snprintf(iPuff, sizeof(iPuff), jsID2, iMuff, Port, iName);
   SidJSON[2] = strdup(iPuff);
   LidJSON[2] = strlen(iPuff);
 
   strcpy(iPuff+n-1, ",");
   snprintf(iMuff, sizeof(iMuff), jsID3, iPuff, iProg, XrdVERSION);
   SidJSON[3] = strdup(iMuff);
   LidJSON[3] = strlen(iMuff);
}
 
/******************************************************************************/
  
int XrdXrootdMonitor::Init()
{
   static     XrdXrootdMonitor_Ident MonIdent(monIdent, isEnabled);
   int        i, Now = time(0);
   bool       aOK;
 
// Setup the primary destination
//
   if (Dest1)
      {InetDest1 = new XrdNetMsg(eDest, Dest1, &aOK, true);
       if (!aOK)
          {eDest->Emsg("Monitor", "Unable to setup primary monitor collector.");
           return 0;
          }
      }
 
// Setup the secondary destination
//
   if (Dest2)
      {InetDest2 = new XrdNetMsg(eDest, Dest2, &aOK, true);
       if (!aOK)
          {eDest->Emsg("Monitor","Unable to setup secondary monitor collector.");
           return 0;
          }
      }
 
// Now schedule the first identification record
//
   if (Sched && monIdent >= 0) Sched->Schedule((XrdJob *)&MonIdent);
 
// There is nothing more to do unless we have been enabled via Defaults()
//
   if (!isEnabled) return 1;
 
// If there is a destination that is only collecting file events, then
// allocate a global monitor object but don't start the timer just yet.
//
   if ((monMode1 && !(monMode1 & XROOTD_MON_IO))
   ||  (monMode2 && !(monMode2 & XROOTD_MON_IO)))
       if (!(altMon = new XrdXrootdMonitor()) || !altMon->monBuff)
          {if (altMon) {delete altMon; altMon = 0;}
           eDest->Emsg("Monitor","allocate monitor; insufficient storage.");
           return 0;
          }
 
// Turn on the monitoring clock if we need it running all the time
//
   if (monCLOCK) startClock();
 
// If we are monitoring file stats then start that up
//
   if (!Sched || !monFSTAT) monFSTAT = 0;
      else if (!XrdXrootdMonFile::Init()) return 0;
 
// If we are not monitoring redirections, we are done!
//
   if (!monREDR) return 1;
 
// Allocate as many redirection monitors as requested
//
   for (i = 0; i < rdrNum; i++)
       {if (posix_memalign((void **)&rdrMon[i].Buff, getpagesize(),monRlen))
           {eDest->Emsg("Monitor", "Unable to allocate monitor rdr buffer.");
            return 0;
           }
        rdrMon[i].Buff->sID    = mySID;
        rdrMon[i].Buff->sXX[0] = XROOTD_MON_REDSID;
        rdrMon[i].Next = (i ? &rdrMon[i-1] : &rdrMon[0]);
        rdrMon[i].nextEnt = 0;
        rdrMon[i].flushIt = Now + autoFlush;
        rdrMon[i].lastTOD = 0;
       }
   rdrMon[0].Next = &rdrMon[i-1];
   rdrMP = &rdrMon[0];
 
// All done
//
   return 1;
}
 
/******************************************************************************/
/*                             G e t D i c t I D                              */
/******************************************************************************/
  
kXR_unt32 XrdXrootdMonitor::GetDictID(bool hbo)
{
   static XrdSysMutex  seqMutex;
   static unsigned int monSeqID = 1;
   unsigned int        mySeqID;
 
// Assign a unique ID for this entry
//
   seqMutex.Lock();
   mySeqID = monSeqID++;
   seqMutex.UnLock();
 
// Return the ID
//
   if (hbo) return mySeqID;
   return htonl(mySeqID);
}
 
/******************************************************************************/
/* Private:                          M a p                                    */
/******************************************************************************/
  
kXR_unt32 XrdXrootdMonitor::Map(char  code, XrdXrootdMonitor::User &uInfo,
                                const char *path)
{
   XrdXrootdMonMap     map;
   const int           room = sizeof(map.info);
   int                 size, montype;
 
// Copy in the username and path. The record is a fixed-size object, so the
// size must come from what was copied, not from the source. Otherwise Send()
// puts whatever follows the record on the wire. Send() sends only size bytes,
// so the room for each text ends after its null byte, and no padding occurs.
//
   map.dictid = GetDictID();
   size = CopyText(map.info, std::min(uInfo.Len + 1, room),
                   uInfo.Name, uInfo.Len);
   if (path && size < room - 2)
      {int pLen = strlen(path);
       map.info[size++] = '\n';
       size += CopyText(map.info + size, std::min(pLen + 1, room - size),
                        path, pLen);
      }
 
// Fill in the header
//
   size = sizeof(XrdXrootdMonHeader)+sizeof(kXR_int32)+size;
   fillHeader(&map.hdr, code, size);
 
// Route the packet to all destinations that need them
//
        if (code == XROOTD_MON_MAPPATH) montype = XROOTD_MON_PATH;
   else if (code == XROOTD_MON_MAPUSER
        ||  code == XROOTD_MON_MAPTOKN
        ||  code == XROOTD_MON_MAPUEAC) montype = XROOTD_MON_USER;
   else                                 montype = XROOTD_MON_INFO;
   Send(montype, (void *)&map, size);
 
// Return the dictionary id
//
   return map.dictid;
}
  
/******************************************************************************/
/*                                  O p e n                                   */
/******************************************************************************/
  
void XrdXrootdMonitor::Open(kXR_unt32 dictid, off_t fsize)
{
  XrdXrootdMonitorLock mLock(this);
 
  if (lastWindow != currWindow) Mark();
     else if (nextEnt == lastEnt) Flush();
  h2nll(fsize, monBuff->info[nextEnt].arg0.val);
  monBuff->info[nextEnt].arg0.id[0]    = XROOTD_MON_OPEN;
  monBuff->info[nextEnt].arg1.buflen   = 0;
  monBuff->info[nextEnt++].arg2.dictid = dictid;
 
// Check if we need to duplicate this entry
//
   if (altMon && this != altMon) altMon->Dup(&monBuff->info[nextEnt-1]);
}
 
/******************************************************************************/
/*                              R e d i r e c t                               */
/******************************************************************************/
  
int XrdXrootdMonitor::Redirect(kXR_unt32 mID, const char *hName, int Port,
                               char      opC, const char *Path)
{
   XrdXrootdMonRedir *mtP;
   MonRdrBuff        *mP = Fetch();
   int n, slots, hLen, pLen, room;
   char *dest;
 
// Take care of the server's name which might actually be a path
//
   if (*hName == '/') {Path = hName; hName = ""; hLen = 0;}
      else {const char *quest = index(hName, '?');
            hLen = (quest ? quest - hName : strlen(hName));
            if (hLen >  256) hLen =  256;
           }
 
// Take care of the path. Only the truncated length may be copied, as the
// path can be longer than the slots that hold it. CopyText() then ends the
// path with a null byte and null pads the rest of the slots.
//
   pLen = strlen(Path);
   if (pLen > 1024) pLen = 1024;
 
// Compute number of entries needed here
//
   n = (hLen + 1 + pLen + 1);    // "<host>:<path>\0"
   slots = n / sizeof(XrdXrootdMonRedir);
   if (n % sizeof(XrdXrootdMonRedir)) slots++;
   room = slots * sizeof(XrdXrootdMonRedir) - (hLen+1);
 
// Obtain a lock on this buffer
//
   if (!mP) return 0;
   mP->Mutex.Lock();
 
// If we don't have enough slots, flush this buffer. Note that we account for
// the ending timing mark here (an extra slot).
//
   if (mP->nextEnt + slots + 2 >= lastRnt) Flush(mP);
 
// Check if we need a timing mark
//
   if (mP->lastTOD != rdrTOD)
      {mP->lastTOD = rdrTOD;
       setTMurk(mP->Buff, mP->nextEnt, mP->lastTOD);
       mP->nextEnt++;
      }
 
// Fill out the buffer
//
   mtP = &(mP->Buff->info[mP->nextEnt]);
   mtP->arg0.rdr.Type = XROOTD_MON_REDIRECT | opC;
   mtP->arg0.rdr.Dent = static_cast<char>(slots);
   mtP->arg0.rdr.Port = htons(static_cast<short>(Port));
   mtP->arg1.dictid   = mID;
   dest = (char *)(mtP+1);
   strncpy(dest, hName,hLen); dest += hLen; *dest++ = ':';
   CopyText(dest, room, Path, pLen);
 
// Adjust pointer and return
//
   mP->nextEnt = mP->nextEnt + (slots+1);
   mP->Mutex.UnLock();
   return 0;
}
 
 
/******************************************************************************/
/*                                  T i c k                                   */
/******************************************************************************/
  
time_t XrdXrootdMonitor::Tick()
{
   time_t Now = time(0);
   int    nextFlush;
 
// We can safely set the window as we are the only ones doing so and memory
// access is atomic as long as it sits within a cache line (which it does).
//
   currWindow = static_cast<kXR_int32>(Now);
   rdrTOD     = htonl(currWindow);
   nextFlush  = currWindow + autoFlush;
 
// Check to see if we should flush the alternate monitor
//
   if (altMon && currWindow >= FlushTime)
      {XrdXrootdMonitorLock::Lock();
       if (currWindow >= FlushTime)
          {if (altMon->nextEnt > 1) altMon->Flush();
              else FlushTime = nextFlush;
          }
       XrdXrootdMonitorLock::UnLock();
      }
 
// Now check to see if we need to flush redirect buffers
//
   if (monREDR)
      {int n = rdrNum;
       while(n--)
            {rdrMon[n].Mutex.Lock();
             if (rdrMon[n].nextEnt == 0) rdrMon[n].flushIt = nextFlush;
                else if (rdrMon[n].flushIt <= currWindow) Flush(&rdrMon[n]);
             rdrMon[n].Mutex.UnLock();
            }
      }
 
// All done. Stop the clock if there is no reason for it to be running. The
// clock always runs if we are monitoring redirects or all clients. Otherwise,
// the clock only runs if we have a one or more client-specific monitors.
//
   if (!monREDR && isEnabled < 0)
      {windowMutex.Lock();
       if (!numMonitor) Now = 0;
       windowMutex.UnLock();
      }
   return Now;
}
 
/******************************************************************************/
/*                               u n A l l o c                                */
/******************************************************************************/
  
void XrdXrootdMonitor::unAlloc(XrdXrootdMonitor *monp)
{
 
// We must delete this object if we are de-allocating the local monitor.
//
   if (monp != altMon) delete monp;
 
// Decrease number being monitored if in selective mode
//
   if (isEnabled < 0)
      {windowMutex.Lock();
       numMonitor--;
       windowMutex.UnLock();
      }
}
 
/******************************************************************************/
/*                       P r i v a t e   M e t h o d s                        */
/******************************************************************************/
/******************************************************************************/
/*                              d o _ S h i f t                               */
/******************************************************************************/
  
unsigned char XrdXrootdMonitor::do_Shift(long long xTot, unsigned int &xVal)
{
  const long long smask = 0x7fffffff00000000LL;
  const long long xmask = 0x7fffffffffffffffLL;
  unsigned char xshift = 0;
 
  xTot &= xmask;
  while(xTot & smask) {xTot = xTot >> 1LL; xshift++;}
  xVal = static_cast<unsigned int>(xTot);
 
  return xshift;
}
 
/******************************************************************************/
/*                            f i l l H e a d e r                             */
/******************************************************************************/
  
void XrdXrootdMonitor::fillHeader(XrdXrootdMonHeader *hdr,
                                  const char          id, int size)
{
 
// Fill in the header
//
   hdr->code = static_cast<kXR_char>(id);
// hdr->pseq = static_cast<kXR_char>(myseq);  // Filled in Send()
   hdr->plen = htons(static_cast<uint16_t>(size));
   hdr->stod = startTime;
}
  
/******************************************************************************/
/*                                 F l u s h                                  */
/******************************************************************************/
  
void XrdXrootdMonitor::Flush()
{
   int       size;
   kXR_int32 localWindow, now;
 
// Do not flush if the buffer is empty
//
   if (nextEnt <= 1) return;
 
// Get the current window marker. No need for locks as simple memory accesses
// are sufficiently synchrnozed for our purposes.
//
   localWindow = currWindow;
 
// Fill in the header and in the process we will have the current time
//
   size = (nextEnt+1)*sizeof(XrdXrootdMonTrace)+sizeof(XrdXrootdMonHeader);
   fillHeader(&monBuff->hdr, XROOTD_MON_MAPTRCE, size);
 
// Punt on the right ending time. We are trying to keep same-sized windows
// This was corrected by Matevz Tadel, as before we were using real time which
// could have been far into the future due to simple inactivity. So, Place the
// computed ending timing mark.
//
   now = lastWindow + sizeWindow;
   setTMark(monBuff, nextEnt, now);
 
// Send off the buffer and reinitialize it
//
   if (this != altMon) Send(XROOTD_MON_IO, (void *)monBuff, size);
      else {Send(XROOTD_MON_FILE, (void *)monBuff, size);
            FlushTime = localWindow + autoFlush;
           }
   setTMark(monBuff, 0, localWindow);
   nextEnt = 1;
}
 
/******************************************************************************/
 
void XrdXrootdMonitor::Flush(XrdXrootdMonitor::MonRdrBuff *mP)
{
   int size;
 
// Reset flush time but do not flush an empty buffer. We use the current time
// to make sure a record atleast sits in the buffer a full flush period.
//
   mP->flushIt = static_cast<int>(time(0)) + autoFlush;
   if (mP->nextEnt <= 1) return;
 
// Set ending timing mark and force a new one on the next fill
//
   setTMurk(mP->Buff, mP->nextEnt, rdrTOD);
   mP->lastTOD = 0;
 
// Fill in the header and in the process we will have the current time
//
   size = (mP->nextEnt+1)*sizeof(XrdXrootdMonRedir)+sizeof(XrdXrootdMonHeader)+8;
   fillHeader(&(mP->Buff->hdr), XROOTD_MON_MAPREDR, size);
 
// Send off the buffer and reinitialize it
//
   Send(XROOTD_MON_REDR, (void *)(mP->Buff), size);
   mP->nextEnt = 0;
}
 
/******************************************************************************/
/*                                  M a r k                                   */
/******************************************************************************/
  
void XrdXrootdMonitor::Mark()
{
   kXR_int32 localWindow;
 
// Get the current window marker. Since simple memory accesses are sufficiently
// synchronized, no need to lock this.
//
   localWindow = currWindow;
 
// Using an update provided by Matevz Tadel, UCSD, if this is an I/O buffer
// mark then we will also flush the I/O buffer if all the following hold:
// a) flushing enabled, b) buffer not empty, and c) covers the flush time.
// We would normally do this during Tick() but that would require too much
// locking in the middle of an I/O path, so we do psudo timed flushing.
//
   if (this != altMon && autoFlash && nextEnt > 1)
      {kXR_int32 bufStartWindow = 
                 static_cast<kXR_int32>(ntohl(monBuff->info[0].arg2.Window));
       if (localWindow - bufStartWindow >= autoFlash)
          {Flush();
           lastWindow = localWindow;
           return;
          }
      }
 
// Now, optimize placing the window mark in the buffer. Using another MT fix we
// set the end of the previous window to be lastwindow + sizeWindow (instead of
// localWindow) to prevent windows from being wrongly zero sized.
//
        if (monBuff->info[nextEnt-1].arg0.id[0] == XROOTD_MON_WINDOW)
   {
       monBuff->info[nextEnt-1].arg2.Window =
                static_cast<kXR_int32>(htonl(localWindow));
   }
   else if (nextEnt+8 > lastEnt)
   {
      Flush();
   }
   else
   {
      monBuff->info[nextEnt].arg0.val     = mySID;
      monBuff->info[nextEnt].arg0.id[0]   = XROOTD_MON_WINDOW;
      monBuff->info[nextEnt].arg1.Window  =
               static_cast<kXR_int32>(htonl(lastWindow + sizeWindow));
      monBuff->info[nextEnt].arg2.Window  =
               static_cast<kXR_int32>(htonl(localWindow));
      nextEnt++;
   }
   lastWindow = localWindow;
}
 
/******************************************************************************/
/*                                  S e n d                                   */
/******************************************************************************/
  
int XrdXrootdMonitor::Send(int monMode, void *buff, int blen, bool setseq)
{
#ifndef NODEBUG
    const char *TraceID = "Monitor";
#endif
    static XrdSysMutex sendMutex;
    static int seq1=0, seq2=0;
    XrdXrootdMonHeader *mHdr=0;
    int rc1, rc2;
 
// If we are to set sequence numbers, recast the buffer. We are assured that
// the buffer always starts with the standard monitor header.
//
   if (setseq) mHdr = static_cast<XrdXrootdMonHeader*>(buff);
 
    sendMutex.Lock();
    if (monMode & monMode1 && InetDest1)
       {if (mHdr) mHdr->pseq = (seq1++) & 0xff;
        rc1  = InetDest1->Send((char *)buff, blen);
        TRACE(DEBUG,blen <<" bytes sent to " <<Dest1 <<" rc=" <<rc1);
       }
       else rc1 = 0;
    if (monMode & monMode2 && InetDest2)
       {if (mHdr) mHdr->pseq = (seq2++) & 0xff;
        rc2  = InetDest2->Send((char *)buff, blen);
        TRACE(DEBUG,blen <<" bytes sent to " <<Dest2 <<" rc=" <<rc2);
       }
       else rc2 = 0;
    sendMutex.UnLock();
 
    return (rc1 ? rc1 : rc2);
}
 
/******************************************************************************/
/*                            s t a r t C l o c k                             */
/******************************************************************************/
  
void XrdXrootdMonitor::startClock()
{
   static XrdXrootdMonitor_Tick MonTick;
          time_t Now;
 
// Start the clock, Caller must have windowMutex locked, if necessary.
//
   Now = time(0);
   currWindow = static_cast<kXR_int32>(Now);
   rdrTOD     = htonl(currWindow);
   MonTick.Set(Sched, sizeWindow);
   FlushTime = autoFlush + currWindow;
   if (Sched) Sched->Schedule((XrdJob *)&MonTick, Now+sizeWindow);
}