src/server/core/datacoll.cpp

   1 /*
   2 ** NetXMS - Network Management System
   3 ** Copyright (C) 2003-2016 Victor Kirhenshtein
   4 **
   5 ** This program is free software; you can redistribute it and/or modify
   6 ** it under the terms of the GNU General Public License as published by
   7 ** the Free Software Foundation; either version 2 of the License, or
   8 ** (at your option) any later version.
   9 **
  10 ** This program is distributed in the hope that it will be useful,
  11 ** but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 ** MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 ** GNU General Public License for more details.
  14 **
  15 ** You should have received a copy of the GNU General Public License
  16 ** along with this program; if not, write to the Free Software
  17 ** Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  18 **
  19 ** File: datacoll.cpp
  20 **
  21 **/
  22
  23 #include "nxcore.h"
  24
  25 /**
  26  * Interval between DCI polling
  27  */
  28 #define ITEM_POLLING_INTERVAL    1
  29
  30 /**
  31  * Externals
  32  */
  33 extern Queue g_syslogProcessingQueue;
  34 extern Queue g_syslogWriteQueue;
  35
  36 /**
  37  * Global data
  38  */
  39 double g_dAvgPollerQueueSize = 0;
  40 double g_dAvgDBWriterQueueSize = 0;
  41 double g_dAvgIDataWriterQueueSize = 0;
  42 double g_dAvgRawDataWriterQueueSize = 0;
  43 double g_dAvgDBAndIDataWriterQueueSize = 0;
  44 double g_dAvgSyslogProcessingQueueSize = 0;
  45 double g_dAvgSyslogWriterQueueSize = 0;
  46 UINT32 g_dwAvgDCIQueuingTime = 0;
  47 Queue g_dataCollectionQueue(4096, 256);
  48 Queue g_dciCacheLoaderQueue;
  49
  50 /**
  51  * Collect data for DCI
  52  */
  53 static void *GetItemData(DataCollectionTarget *dcTarget, DCItem *pItem, TCHAR *pBuffer, UINT32 *error)
  54 {
  55    if (dcTarget->getObjectClass() == OBJECT_CLUSTER)
  56    {
  57       if (pItem->isAggregateOnCluster())
  58       {
  59          *error = ((Cluster *)dcTarget)->collectAggregatedData(pItem, pBuffer);
  60       }
  61       else
  62       {
  63          *error = DCE_IGNORE;
  64       }
  65    }
  66    else
  67    {
  68       switch(pItem->getDataSource())
  69       {
  70          case DS_INTERNAL:    // Server internal parameters (like status)
  71             *error = dcTarget->getInternalItem(pItem->getName(), MAX_LINE_SIZE, pBuffer);
  72             break;
  73          case DS_SNMP_AGENT:
  74                            if (dcTarget->getObjectClass() == OBJECT_NODE)
  75                                    *error = ((Node *)dcTarget)->getItemFromSNMP(pItem->getSnmpPort(), pItem->getName(), MAX_LINE_SIZE,
  76                                            pBuffer, pItem->isInterpretSnmpRawValue() ? (int)pItem->getSnmpRawValueType() : SNMP_RAWTYPE_NONE);
  77                            else
  78                                    *error = DCE_NOT_SUPPORTED;
  79             break;
  80          case DS_CHECKPOINT_AGENT:
  81                            if (dcTarget->getObjectClass() == OBJECT_NODE)
  82                     *error = ((Node *)dcTarget)->getItemFromCheckPointSNMP(pItem->getName(), MAX_LINE_SIZE, pBuffer);
  83                            else
  84                                    *error = DCE_NOT_SUPPORTED;
  85             break;
  86          case DS_NATIVE_AGENT:
  87                            if (dcTarget->getObjectClass() == OBJECT_NODE)
  88                     *error = ((Node *)dcTarget)->getItemFromAgent(pItem->getName(), MAX_LINE_SIZE, pBuffer);
  89                            else
  90                                    *error = DCE_NOT_SUPPORTED;
  91             break;
  92          case DS_WINPERF:
  93                            if (dcTarget->getObjectClass() == OBJECT_NODE)
  94                            {
  95                                    TCHAR name[MAX_PARAM_NAME];
  96                                    _sntprintf(name, MAX_PARAM_NAME, _T("PDH.CounterValue(\"%s\",%d)"), pItem->getName(), pItem->getSampleCount());
  97                     *error = ((Node *)dcTarget)->getItemFromAgent(name, MAX_LINE_SIZE, pBuffer);
  98                            }
  99                            else
 100                            {
 101                                    *error = DCE_NOT_SUPPORTED;
 102                            }
 103             break;
 104          case DS_SMCLP:
 105             if (dcTarget->getObjectClass() == OBJECT_NODE)
 106             {
 107                     *error = ((Node *)dcTarget)->getItemFromSMCLP(pItem->getName(), MAX_LINE_SIZE, pBuffer);
 108             }
 109             else
 110             {
 111                *error = DCE_NOT_SUPPORTED;
 112             }
 113             break;
 114          case DS_SCRIPT:
 115             *error = dcTarget->getScriptItem(pItem->getName(), MAX_LINE_SIZE, pBuffer);
 116             break;
 117                    default:
 118                            *error = DCE_NOT_SUPPORTED;
 119                            break;
 120       }
 121    }
 122         return pBuffer;
 123 }
 124
 125 /**
 126  * Collect data for table
 127  */
 128 static void *GetTableData(DataCollectionTarget *dcTarget, DCTable *table, UINT32 *error)
 129 {
 130         Table *result = NULL;
 131    if (dcTarget->getObjectClass() == OBJECT_CLUSTER)
 132    {
 133       if (table->isAggregateOnCluster())
 134       {
 135          *error = ((Cluster *)dcTarget)->collectAggregatedData(table, &result);
 136       }
 137       else
 138       {
 139          *error = DCE_IGNORE;
 140       }
 141    }
 142    else
 143    {
 144       switch(table->getDataSource())
 145       {
 146                    case DS_NATIVE_AGENT:
 147                            if (dcTarget->getObjectClass() == OBJECT_NODE)
 148             {
 149                                    *error = ((Node *)dcTarget)->getTableFromAgent(table->getName(), &result);
 150                if ((*error == DCE_SUCCESS) && (result != NULL))
 151                   table->updateResultColumns(result);
 152             }
 153                            else
 154             {
 155                                    *error = DCE_NOT_SUPPORTED;
 156             }
 157                            break;
 158          case DS_SNMP_AGENT:
 159                            if (dcTarget->getObjectClass() == OBJECT_NODE)
 160             {
 161                *error = ((Node *)dcTarget)->getTableFromSNMP(table->getSnmpPort(), table->getName(), table->getColumns(), &result);
 162                if ((*error == DCE_SUCCESS) && (result != NULL))
 163                   table->updateResultColumns(result);
 164             }
 165                            else
 166             {
 167                                    *error = DCE_NOT_SUPPORTED;
 168             }
 169             break;
 170                    default:
 171                            *error = DCE_NOT_SUPPORTED;
 172                            break;
 173            }
 174    }
 175         return result;
 176 }
 177
 178 /**
 179  * Data collector
 180  */
 181 static THREAD_RESULT THREAD_CALL DataCollector(void *pArg)
 182 {
 183    UINT32 dwError;
 184
 185    TCHAR *pBuffer = (TCHAR *)malloc(MAX_LINE_SIZE * sizeof(TCHAR));
 186    while(!IsShutdownInProgress())
 187    {
 188       DCObject *pItem = (DCObject *)g_dataCollectionQueue.getOrBlock();
 189                 DataCollectionTarget *target = (DataCollectionTarget *)pItem->getOwner();
 190
 191                 if (pItem->isScheduledForDeletion())
 192                 {
 193               nxlog_debug(7, _T("DataCollector(): about to destroy DC object %d \"%s\" owner=%d"),
 194                                     pItem->getId(), pItem->getName(), (target != NULL) ? (int)target->getId() : -1);
 195                         pItem->deleteFromDatabase();
 196                         delete pItem;
 197                         continue;
 198                 }
 199
 200                 if (target == NULL)
 201                 {
 202          nxlog_debug(3, _T("DataCollector: attempt to collect information for non-existing node (DCI=%d \"%s\")"),
 203                      pItem->getId(), pItem->getName());
 204
 205          // Update item's last poll time and clear busy flag so item can be polled again
 206          pItem->setLastPollTime(time(NULL));
 207          pItem->setBusyFlag(FALSE);
 208                    continue;
 209                 }
 210
 211       DbgPrintf(8, _T("DataCollector(): processing DC object %d \"%s\" owner=%d sourceNode=%d"),
 212                           pItem->getId(), pItem->getName(), (target != NULL) ? (int)target->getId() : -1, pItem->getSourceNode());
 213       UINT32 sourceNodeId = target->getEffectiveSourceNode(pItem);
 214                 if (sourceNodeId != 0)
 215                 {
 216                         Node *sourceNode = (Node *)FindObjectById(sourceNodeId, OBJECT_NODE);
 217                         if (sourceNode != NULL)
 218                         {
 219                                 if (((target->getObjectClass() == OBJECT_CHASSIS) && (((Chassis *)target)->getControllerId() == sourceNodeId)) ||
 220                                     sourceNode->isTrustedNode(target->getId()))
 221                                 {
 222                                         target = sourceNode;
 223                                         target->incRefCount();
 224                                 }
 225                                 else
 226                                 {
 227                // Change item's status to _T("not supported")
 228                pItem->setStatus(ITEM_STATUS_NOT_SUPPORTED, true);
 229                target->decRefCount();
 230                target = NULL;
 231                                 }
 232                         }
 233                         else
 234                         {
 235             target->decRefCount();
 236             target = NULL;
 237                         }
 238                 }
 239
 240       time_t currTime = time(NULL);
 241       if (target != NULL)
 242       {
 243          if (!IsShutdownInProgress())
 244          {
 245             void *data;
 246             switch(pItem->getType())
 247             {
 248                case DCO_TYPE_ITEM:
 249                   data = GetItemData(target, (DCItem *)pItem, pBuffer, &dwError);
 250                   break;
 251                case DCO_TYPE_TABLE:
 252                   data = GetTableData(target, (DCTable *)pItem, &dwError);
 253                   break;
 254                default:
 255                   data = NULL;
 256                   dwError = DCE_NOT_SUPPORTED;
 257                   break;
 258             }
 259
 260             // Transform and store received value into database or handle error
 261             switch(dwError)
 262             {
 263                case DCE_SUCCESS:
 264                   if (pItem->getStatus() == ITEM_STATUS_NOT_SUPPORTED)
 265                      pItem->setStatus(ITEM_STATUS_ACTIVE, true);
 266                   if (!((DataCollectionTarget *)pItem->getOwner())->processNewDCValue(pItem, currTime, data))
 267                   {
 268                      // value processing failed, convert to data collection error
 269                      pItem->processNewError(false);
 270                   }
 271                   break;
 272                case DCE_COMM_ERROR:
 273                   if (pItem->getStatus() == ITEM_STATUS_NOT_SUPPORTED)
 274                      pItem->setStatus(ITEM_STATUS_ACTIVE, true);
 275                   pItem->processNewError(false);
 276                   break;
 277                case DCE_NO_SUCH_INSTANCE:
 278                   pItem->processNewError(true);
 279                   break;
 280                case DCE_NOT_SUPPORTED:
 281                   // Change item's status
 282                   pItem->setStatus(ITEM_STATUS_NOT_SUPPORTED, true);
 283                   break;
 284             }
 285          }
 286
 287          // Decrement node's usage counter
 288          target->decRefCount();
 289                         if ((pItem->getSourceNode() != 0) && (pItem->getOwner() != NULL))
 290                         {
 291                                 pItem->getOwner()->decRefCount();
 292                         }
 293       }
 294       else     /* target == NULL */
 295       {
 296                         Template *n = pItem->getOwner();
 297          nxlog_debug(5, _T("DataCollector: attempt to collect information for non-existing or inaccessible node (DCI=%d \"%s\" target=%d sourceNode=%d)"),
 298                                     pItem->getId(), pItem->getName(), (n != NULL) ? (int)n->getId() : -1, sourceNodeId);
 299       }
 300
 301                 // Update item's last poll time and clear busy flag so item can be polled again
 302       pItem->setLastPollTime(currTime);
 303       pItem->setBusyFlag(FALSE);
 304    }
 305
 306    free(pBuffer);
 307    DbgPrintf(1, _T("Data collector thread terminated"));
 308    return THREAD_OK;
 309 }
 310
 311 /**
 312  * Callback for queueing DCIs
 313  */
 314 static void QueueItems(NetObj *object, void *data)
 315 {
 316    if (IsShutdownInProgress())
 317       return;
 318
 319         DbgPrintf(8, _T("ItemPoller: calling DataCollectionTarget::queueItemsForPolling for object %s [%d]"),
 320                                  object->getName(), object->getId());
 321         ((DataCollectionTarget *)object)->queueItemsForPolling(&g_dataCollectionQueue);
 322 }
 323
 324 /**
 325  * Item poller thread: check nodes' items and put into the
 326  * data collector queue when data polling required
 327  */
 328 static THREAD_RESULT THREAD_CALL ItemPoller(void *pArg)
 329 {
 330    UINT32 dwSum, dwWatchdogId, currPos = 0;
 331    UINT32 dwTimingHistory[60 / ITEM_POLLING_INTERVAL];
 332    INT64 qwStart;
 333
 334    dwWatchdogId = WatchdogAddThread(_T("Item Poller"), 20);
 335    memset(dwTimingHistory, 0, sizeof(UINT32) * (60 / ITEM_POLLING_INTERVAL));
 336
 337    while(!IsShutdownInProgress())
 338    {
 339       if (SleepAndCheckForShutdown(ITEM_POLLING_INTERVAL))
 340          break;      // Shutdown has arrived
 341       WatchdogNotify(dwWatchdogId);
 342                 DbgPrintf(8, _T("ItemPoller: wakeup"));
 343
 344       qwStart = GetCurrentTimeMs();
 345                 g_idxNodeById.forEach(QueueItems, NULL);
 346                 g_idxClusterById.forEach(QueueItems, NULL);
 347                 g_idxMobileDeviceById.forEach(QueueItems, NULL);
 348       g_idxChassisById.forEach(QueueItems, NULL);
 349
 350       // Save last poll time
 351       dwTimingHistory[currPos] = (UINT32)(GetCurrentTimeMs() - qwStart);
 352       currPos++;
 353       if (currPos == (60 / ITEM_POLLING_INTERVAL))
 354          currPos = 0;
 355
 356       // Calculate new average for last minute
 357                 dwSum = 0;
 358       for(int i = 0; i < (60 / ITEM_POLLING_INTERVAL); i++)
 359          dwSum += dwTimingHistory[i];
 360       g_dwAvgDCIQueuingTime = dwSum / (60 / ITEM_POLLING_INTERVAL);
 361    }
 362    DbgPrintf(1, _T("Item poller thread terminated"));
 363    return THREAD_OK;
 364 }
 365
 366 /**
 367  * Statistics collection thread
 368  */
 369 static THREAD_RESULT THREAD_CALL StatCollector(void *pArg)
 370 {
 371    UINT32 i, currPos = 0;
 372    UINT32 pollerQS[12], dbWriterQS[12];
 373    UINT32 iDataWriterQS[12], rawDataWriterQS[12], dbAndIDataWriterQS[12];
 374    UINT32 syslogProcessingQS[12], syslogWriterQS[12];
 375    double sum1, sum2, sum3, sum4, sum5, sum8, sum9;
 376
 377    memset(pollerQS, 0, sizeof(UINT32) * 12);
 378    memset(dbWriterQS, 0, sizeof(UINT32) * 12);
 379    memset(iDataWriterQS, 0, sizeof(UINT32) * 12);
 380    memset(rawDataWriterQS, 0, sizeof(UINT32) * 12);
 381    memset(dbAndIDataWriterQS, 0, sizeof(UINT32) * 12);
 382    memset(syslogProcessingQS, 0, sizeof(UINT32) * 12);
 383    memset(syslogWriterQS, 0, sizeof(UINT32) * 12);
 384    g_dAvgPollerQueueSize = 0;
 385    g_dAvgDBWriterQueueSize = 0;
 386    g_dAvgIDataWriterQueueSize = 0;
 387    g_dAvgRawDataWriterQueueSize = 0;
 388    g_dAvgDBAndIDataWriterQueueSize = 0;
 389    g_dAvgSyslogProcessingQueueSize = 0;
 390    g_dAvgSyslogWriterQueueSize = 0;
 391    while(!IsShutdownInProgress())
 392    {
 393       if (SleepAndCheckForShutdown(5))
 394          break;      // Shutdown has arrived
 395
 396       // Get current values
 397       pollerQS[currPos] = g_dataCollectionQueue.size();
 398       dbWriterQS[currPos] = g_dbWriterQueue->size();
 399       iDataWriterQS[currPos] = g_dciDataWriterQueue->size();
 400       rawDataWriterQS[currPos] = g_dciRawDataWriterQueue->size();
 401       dbAndIDataWriterQS[currPos] = g_dbWriterQueue->size() + g_dciDataWriterQueue->size() + g_dciRawDataWriterQueue->size();
 402       syslogProcessingQS[currPos] = g_syslogProcessingQueue.size();
 403       syslogWriterQS[currPos] = g_syslogWriteQueue.size();
 404       currPos++;
 405       if (currPos == 12)
 406          currPos = 0;
 407
 408       // Calculate new averages
 409       for(i = 0, sum1 = 0, sum2 = 0, sum3 = 0, sum4 = 0, sum5 = 0, sum8 = 0, sum9 = 0; i < 12; i++)
 410       {
 411          sum1 += pollerQS[i];
 412          sum2 += dbWriterQS[i];
 413          sum3 += iDataWriterQS[i];
 414          sum4 += rawDataWriterQS[i];
 415          sum5 += dbAndIDataWriterQS[i];
 416          sum8 += syslogProcessingQS[i];
 417          sum9 += syslogWriterQS[i];
 418       }
 419       g_dAvgPollerQueueSize = sum1 / 12;
 420       g_dAvgDBWriterQueueSize = sum2 / 12;
 421       g_dAvgIDataWriterQueueSize = sum3 / 12;
 422       g_dAvgRawDataWriterQueueSize = sum4 / 12;
 423       g_dAvgDBAndIDataWriterQueueSize = sum5 / 12;
 424       g_dAvgSyslogProcessingQueueSize = sum8 / 12;
 425       g_dAvgSyslogWriterQueueSize = sum9 / 12;
 426    }
 427    return THREAD_OK;
 428 }
 429
 430 /**
 431  * DCI cache loader
 432  */
 433 THREAD_RESULT THREAD_CALL CacheLoader(void *arg)
 434 {
 435    DbgPrintf(2, _T("DCI cache loader thread started"));
 436    while(true)
 437    {
 438       DCItem *dci = (DCItem *)g_dciCacheLoaderQueue.getOrBlock();
 439       if (dci == INVALID_POINTER_VALUE)
 440          break;
 441
 442       DbgPrintf(6, _T("Loading cache for DCI %s [%d] on %s [%d]"),
 443                 dci->getName(), dci->getId(), dci->getOwnerName(), dci->getOwnerId());
 444       dci->reloadCache();
 445       dci->getOwner()->decRefCount();
 446    }
 447    DbgPrintf(2, _T("DCI cache loader thread stopped"));
 448    return THREAD_OK;
 449 }
 450
 451 /**
 452  * Initialize data collection subsystem
 453  */
 454 BOOL InitDataCollector()
 455 {
 456    int i, iNumCollectors;
 457
 458    // Start data collection threads
 459    iNumCollectors = ConfigReadInt(_T("NumberOfDataCollectors"), 10);
 460    for(i = 0; i < iNumCollectors; i++)
 461       ThreadCreate(DataCollector, 0, NULL);
 462
 463    ThreadCreate(ItemPoller, 0, NULL);
 464    ThreadCreate(StatCollector, 0, NULL);
 465    ThreadCreate(CacheLoader, 0, NULL);
 466
 467    return TRUE;
 468 }
 469
 470 /**
 471  * Update parameter list from node
 472  */
 473 static void UpdateParamList(NetObj *object, void *data)
 474 {
 475         ObjectArray<AgentParameterDefinition> *fullList = (ObjectArray<AgentParameterDefinition> *)data;
 476
 477         ObjectArray<AgentParameterDefinition> *paramList;
 478         ((Node *)object)->openParamList(&paramList);
 479         if ((paramList != NULL) && (paramList->size() > 0))
 480         {
 481                 for(int i = 0; i < paramList->size(); i++)
 482                 {
 483                         int j;
 484                         for(j = 0; j < fullList->size(); j++)
 485                         {
 486                                 if (!_tcsicmp(paramList->get(i)->getName(), fullList->get(j)->getName()))
 487                                         break;
 488                         }
 489
 490                         if (j == fullList->size())
 491                         {
 492             fullList->add(new AgentParameterDefinition(paramList->get(i)));
 493                         }
 494                 }
 495         }
 496         ((Node *)object)->closeParamList();
 497 }
 498
 499 /**
 500  * Update table list from node
 501  */
 502 static void UpdateTableList(NetObj *object, void *data)
 503 {
 504         ObjectArray<AgentTableDefinition> *fullList = (ObjectArray<AgentTableDefinition> *)data;
 505
 506    ObjectArray<AgentTableDefinition> *tableList;
 507         ((Node *)object)->openTableList(&tableList);
 508         if ((tableList != NULL) && (tableList->size() > 0))
 509         {
 510                 for(int i = 0; i < tableList->size(); i++)
 511                 {
 512                         int j;
 513                         for(j = 0; j < fullList->size(); j++)
 514                         {
 515                                 if (!_tcsicmp(tableList->get(i)->getName(), fullList->get(j)->getName()))
 516                                         break;
 517                         }
 518
 519                         if (j == fullList->size())
 520                         {
 521             fullList->add(new AgentTableDefinition(tableList->get(i)));
 522                         }
 523                 }
 524         }
 525         ((Node *)object)->closeTableList();
 526 }
 527
 528 /**
 529  * Write full (from all nodes) agent parameters list to NXCP message
 530  */
 531 void WriteFullParamListToMessage(NXCPMessage *pMsg, WORD flags)
 532 {
 533    // Gather full parameter list
 534         if (flags & 0x01)
 535         {
 536                 ObjectArray<AgentParameterDefinition> fullList(64, 64, true);
 537                 g_idxNodeById.forEach(UpdateParamList, &fullList);
 538
 539                 // Put list into the message
 540                 pMsg->setField(VID_NUM_PARAMETERS, (UINT32)fullList.size());
 541       UINT32 varId = VID_PARAM_LIST_BASE;
 542                 for(int i = 0; i < fullList.size(); i++)
 543                 {
 544          varId += fullList.get(i)->fillMessage(pMsg, varId);
 545                 }
 546         }
 547
 548    // Gather full table list
 549         if (flags & 0x02)
 550         {
 551                 ObjectArray<AgentTableDefinition> fullList(64, 64, true);
 552                 g_idxNodeById.forEach(UpdateTableList, &fullList);
 553
 554                 // Put list into the message
 555                 pMsg->setField(VID_NUM_TABLES, (UINT32)fullList.size());
 556       UINT32 varId = VID_TABLE_LIST_BASE;
 557                 for(int i = 0; i < fullList.size(); i++)
 558                 {
 559          varId += fullList.get(i)->fillMessage(pMsg, varId);
 560                 }
 561         }
 562 }
 563
 564 /**
 565  * Get type of data collection object
 566  */
 567 int GetDCObjectType(UINT32 nodeId, UINT32 dciId)
 568 {
 569    Node *node = (Node *)FindObjectById(nodeId, OBJECT_NODE);
 570    if (node != NULL)
 571    {
 572       DCObject *dco = node->getDCObjectById(dciId);
 573       if (dco != NULL)
 574       {
 575          return dco->getType();
 576       }
 577    }
 578    return DCO_TYPE_ITEM;   // default
 579 }