- Initial DCI transformation support on server side
[public/netxms.git] / src / server / core / dcitem.cpp
CommitLineData
3ea35b38
VK
1/*
2** NetXMS - Network Management System
3** Copyright (C) 2003, 2004 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** $module: dcitem.cpp
20**
21**/
22
23#include "nms_core.h"
24
25
26//
27// Default constructor for DCItem
28//
29
30DCItem::DCItem()
31{
32 m_dwId = 0;
333ece94 33 m_dwTemplateId = 0;
3ea35b38
VK
34 m_dwNumThresholds = 0;
35 m_ppThresholdList = NULL;
36 m_iBusy = 0;
42d7ed00 37 m_iDataType = DCI_DT_INTEGER;
3ea35b38
VK
38 m_iPollingInterval = 3600;
39 m_iRetentionTime = 0;
333ece94 40 m_iDeltaCalculation = DCM_ORIGINAL_VALUE;
3ea35b38
VK
41 m_iSource = DS_INTERNAL;
42 m_iStatus = ITEM_STATUS_NOT_SUPPORTED;
43 m_szName[0] = 0;
44 m_tLastPoll = 0;
333ece94 45 m_pszFormula = strdup("");
3ea35b38 46 m_pNode = NULL;
07a45e04 47 m_hMutex = MutexCreate();
333ece94
VK
48 m_dwCacheSize = 0;
49 m_ppValueCache = NULL;
3ea35b38
VK
50}
51
52
53//
54// Constructor for creating DCItem from database
55// Assumes that fields in SELECT query are in following order:
333ece94
VK
56// item_id,name,source,datatype,polling_interval,retention_time,status,
57// delta_calculation,transformation,template_id
3ea35b38
VK
58//
59
60DCItem::DCItem(DB_RESULT hResult, int iRow, Node *pNode)
61{
62 m_dwId = DBGetFieldULong(hResult, iRow, 0);
63 strcpy(m_szName, DBGetField(hResult, iRow, 1));
64 m_iSource = (BYTE)DBGetFieldLong(hResult, iRow, 2);
65 m_iDataType = (BYTE)DBGetFieldLong(hResult, iRow, 3);
66 m_iPollingInterval = DBGetFieldLong(hResult, iRow, 4);
67 m_iRetentionTime = DBGetFieldLong(hResult, iRow, 5);
68 m_iStatus = (BYTE)DBGetFieldLong(hResult, iRow, 6);
333ece94
VK
69 m_iDeltaCalculation = (BYTE)DBGetFieldLong(hResult, iRow, 7);
70 m_pszFormula = strdup(DBGetField(hResult, iRow, 8));
71 m_dwTemplateId = DBGetFieldULong(hResult, iRow, 9);
3ea35b38
VK
72 m_iBusy = 0;
73 m_tLastPoll = 0;
74 m_dwNumThresholds = 0;
75 m_ppThresholdList = NULL;
76 m_pNode = pNode;
07a45e04 77 m_hMutex = MutexCreate();
333ece94
VK
78 m_dwCacheSize = 0;
79 m_ppValueCache = NULL;
3ea35b38
VK
80}
81
82
83//
84// Constructor for creating new DCItem from scratch
85//
86
87DCItem::DCItem(DWORD dwId, char *szName, int iSource, int iDataType,
88 int iPollingInterval, int iRetentionTime, Node *pNode)
89{
90 m_dwId = dwId;
333ece94 91 m_dwTemplateId = 0;
3ea35b38
VK
92 strncpy(m_szName, szName, MAX_ITEM_NAME);
93 m_iSource = iSource;
94 m_iDataType = iDataType;
95 m_iPollingInterval = iPollingInterval;
96 m_iRetentionTime = iRetentionTime;
333ece94 97 m_iDeltaCalculation = DCM_ORIGINAL_VALUE;
3ea35b38
VK
98 m_iStatus = ITEM_STATUS_ACTIVE;
99 m_iBusy = 0;
100 m_tLastPoll = 0;
333ece94 101 m_pszFormula = strdup("");
3ea35b38
VK
102 m_dwNumThresholds = 0;
103 m_ppThresholdList = NULL;
104 m_pNode = pNode;
07a45e04 105 m_hMutex = MutexCreate();
333ece94
VK
106 m_dwCacheSize = 0;
107 m_ppValueCache = NULL;
3ea35b38
VK
108}
109
110
111//
112// Destructor for DCItem
113//
114
115DCItem::~DCItem()
116{
117 DWORD i;
118
119 for(i = 0; i < m_dwNumThresholds; i++)
120 delete m_ppThresholdList[i];
121 safe_free(m_ppThresholdList);
333ece94
VK
122 safe_free(m_pszFormula);
123 for(i = 0; i < m_dwCacheSize; i++)
124 delete m_ppValueCache[i];
125 safe_free(m_ppValueCache);
07a45e04 126 MutexDestroy(m_hMutex);
3ea35b38
VK
127}
128
129
3ea35b38
VK
130//
131// Load data collection items thresholds from database
132//
133
134BOOL DCItem::LoadThresholdsFromDB(void)
135{
136 DWORD i;
137 char szQuery[256];
138 DB_RESULT hResult;
139 BOOL bResult = FALSE;
140
141 sprintf(szQuery, "SELECT threshold_id,fire_value,rearm_value,check_function,"
142 "check_operation,parameter_1,parameter_2,event_code FROM thresholds "
06e7be2f 143 "WHERE item_id=%ld ORDER BY sequence_number", m_dwId);
3ea35b38
VK
144 hResult = DBSelect(g_hCoreDB, szQuery);
145 if (hResult != NULL)
146 {
147 m_dwNumThresholds = DBGetNumRows(hResult);
148 if (m_dwNumThresholds > 0)
149 {
150 m_ppThresholdList = (Threshold **)malloc(sizeof(Threshold *) * m_dwNumThresholds);
151 for(i = 0; i < m_dwNumThresholds; i++)
152 m_ppThresholdList[i] = new Threshold(hResult, i, this);
153 }
154 DBFreeResult(hResult);
155 bResult = TRUE;
156 }
157
158 return bResult;
159}
160
161
162//
163// Save to database
164//
165
166BOOL DCItem::SaveToDB(void)
167{
333ece94 168 char *pszEscFormula, szQuery[1024];
3ea35b38
VK
169 DB_RESULT hResult;
170 BOOL bNewObject = TRUE, bResult;
171
07a45e04
VK
172 Lock();
173
3ea35b38
VK
174 // Check for object's existence in database
175 sprintf(szQuery, "SELECT item_id FROM items WHERE item_id=%ld", m_dwId);
176 hResult = DBSelect(g_hCoreDB, szQuery);
177 if (hResult != 0)
178 {
179 if (DBGetNumRows(hResult) > 0)
180 bNewObject = FALSE;
181 DBFreeResult(hResult);
182 }
183
184 // Prepare and execute query
333ece94 185 pszEscFormula = EncodeSQLString(m_pszFormula);
3ea35b38 186 if (bNewObject)
333ece94
VK
187 sprintf(szQuery, "INSERT INTO items (item_id,node_id,template_id,name,description,source,"
188 "datatype,polling_interval,retention_time,status,delta_calculation,"
189 "transformation) VALUES (%ld,%ld,%ld,'%s','',%d,%d,%ld,%ld,%d,%d,'%s')",
190 m_dwId, (m_pNode == NULL) ? 0 : m_pNode->Id(), m_dwTemplateId,
3ea35b38 191 m_szName, m_iSource, m_iDataType, m_iPollingInterval,
333ece94 192 m_iRetentionTime, m_iStatus, m_iDeltaCalculation, pszEscFormula);
3ea35b38 193 else
333ece94
VK
194 sprintf(szQuery, "UPDATE items SET node_id=%ld,template_id=%ld,name='%s',source=%d,"
195 "datatype=%d,polling_interval=%ld,retention_time=%ld,status=%d,"
196 "delta_calculation=%d,transformation='%s' WHERE item_id=%ld",
197 (m_pNode == NULL) ? 0 : m_pNode->Id(), m_dwTemplateId,
198 m_szName, m_iSource, m_iDataType, m_iPollingInterval,
199 m_iRetentionTime, m_iStatus, m_iDeltaCalculation, pszEscFormula, m_dwId);
3ea35b38 200 bResult = DBQuery(g_hCoreDB, szQuery);
333ece94 201 free(pszEscFormula);
3ea35b38
VK
202
203 // Save thresholds
204 if (bResult)
205 {
206 DWORD i;
207
208 for(i = 0; i < m_dwNumThresholds; i++)
06e7be2f
VK
209 m_ppThresholdList[i]->SaveToDB(i);
210 }
211
212 // Delete non-existing thresholds
213 sprintf(szQuery, "SELECT threshold_id FROM thresholds WHERE item_id=%ld", m_dwId);
214 hResult = DBSelect(g_hCoreDB, szQuery);
215 if (hResult != 0)
216 {
217 int i, iNumRows;
218 DWORD j, dwId;
219
220 iNumRows = DBGetNumRows(hResult);
221 for(i = 0; i < iNumRows; i++)
222 {
223 dwId = DBGetFieldULong(hResult, i, 0);
224 for(j = 0; j < m_dwNumThresholds; j++)
225 if (m_ppThresholdList[j]->Id() == dwId)
226 break;
227 if (j == m_dwNumThresholds)
228 {
229 sprintf(szQuery, "DELETE FROM thresholds WHERE threshold_id=%ld", dwId);
230 DBQuery(g_hCoreDB, szQuery);
231 }
232 }
233 DBFreeResult(hResult);
3ea35b38
VK
234 }
235
07a45e04 236 Unlock();
3ea35b38
VK
237 return bResult;
238}
239
240
241//
242// Check last value for threshold breaches
243//
244
333ece94 245void DCItem::CheckThresholds(ItemValue &value)
3ea35b38
VK
246{
247 DWORD i, iResult;
248
07a45e04 249 Lock();
3ea35b38
VK
250 for(i = 0; i < m_dwNumThresholds; i++)
251 {
333ece94 252 iResult = m_ppThresholdList[i]->Check(value);
3ea35b38
VK
253 switch(iResult)
254 {
255 case THRESHOLD_REACHED:
256 PostEvent(m_ppThresholdList[i]->EventCode(), m_pNode->Id(), "sss", m_szName,
333ece94 257 m_ppThresholdList[i]->Value(), (const char *)value);
06e7be2f 258 i = m_dwNumThresholds; // Stop processing
3ea35b38
VK
259 break;
260 case THRESHOLD_REARMED:
261 PostEvent(EVENT_THRESHOLD_REARMED, m_pNode->Id(), "s", m_szName);
262 break;
06e7be2f
VK
263 case NO_ACTION:
264 if (m_ppThresholdList[i]->IsReached())
265 i = m_dwNumThresholds; // Threshold condition still true, stop processing
266 break;
3ea35b38
VK
267 }
268 }
07a45e04 269 Unlock();
3ea35b38 270}
7257eb7d
VK
271
272
273//
274// Create CSCP message with item data
275//
276
277void DCItem::CreateMessage(CSCPMessage *pMsg)
278{
07a45e04
VK
279 DCI_THRESHOLD dct;
280 DWORD i, dwId;
281
282 Lock();
7257eb7d
VK
283 pMsg->SetVariable(VID_DCI_ID, m_dwId);
284 pMsg->SetVariable(VID_NAME, m_szName);
285 pMsg->SetVariable(VID_POLLING_INTERVAL, (DWORD)m_iPollingInterval);
286 pMsg->SetVariable(VID_RETENTION_TIME, (DWORD)m_iRetentionTime);
287 pMsg->SetVariable(VID_DCI_SOURCE_TYPE, (WORD)m_iSource);
288 pMsg->SetVariable(VID_DCI_DATA_TYPE, (WORD)m_iDataType);
289 pMsg->SetVariable(VID_DCI_STATUS, (WORD)m_iStatus);
333ece94
VK
290 pMsg->SetVariable(VID_DCI_DELTA_CALCULATION, (WORD)m_iDeltaCalculation);
291 pMsg->SetVariable(VID_DCI_FORMULA, m_pszFormula);
2b5a7f36 292 pMsg->SetVariable(VID_NUM_THRESHOLDS, m_dwNumThresholds);
07a45e04
VK
293 for(i = 0, dwId = VID_DCI_THRESHOLD_BASE; i < m_dwNumThresholds; i++, dwId++)
294 {
295 m_ppThresholdList[i]->CreateMessage(&dct);
296 pMsg->SetVariable(dwId, (BYTE *)&dct, sizeof(DCI_THRESHOLD));
297 }
298 Unlock();
7257eb7d 299}
9ed4eaff
VK
300
301
302//
303// Delete item and collected data from database
304//
305
306void DCItem::DeleteFromDB(void)
307{
308 char szQuery[256];
309
310 sprintf(szQuery, "DELETE FROM items WHERE item_id=%d", m_dwId);
311 QueueSQLRequest(szQuery);
312 sprintf(szQuery, "DELETE FROM idata_%d WHERE item_id=%d", m_pNode->Id(), m_dwId);
313 QueueSQLRequest(szQuery);
314 sprintf(szQuery, "DELETE FROM thresholds WHERE item_id=%d", m_dwId);
315 QueueSQLRequest(szQuery);
316}
317
318
319//
320// Update item from CSCP message
321//
322
07a45e04
VK
323void DCItem::UpdateFromMessage(CSCPMessage *pMsg, DWORD *pdwNumMaps,
324 DWORD **ppdwMapIndex, DWORD **ppdwMapId)
9ed4eaff 325{
07a45e04
VK
326 DWORD i, j, dwNum, dwId;
327 DCI_THRESHOLD *pNewThresholds;
06e7be2f 328 Threshold **ppNewList;
07a45e04
VK
329
330 Lock();
331
9ed4eaff
VK
332 pMsg->GetVariableStr(VID_NAME, m_szName, MAX_ITEM_NAME);
333 m_iSource = (BYTE)pMsg->GetVariableShort(VID_DCI_SOURCE_TYPE);
334 m_iDataType = (BYTE)pMsg->GetVariableShort(VID_DCI_DATA_TYPE);
335 m_iPollingInterval = pMsg->GetVariableLong(VID_POLLING_INTERVAL);
336 m_iRetentionTime = pMsg->GetVariableLong(VID_RETENTION_TIME);
337 m_iStatus = (BYTE)pMsg->GetVariableShort(VID_DCI_STATUS);
333ece94
VK
338 m_iDeltaCalculation = (BYTE)pMsg->GetVariableShort(VID_DCI_DELTA_CALCULATION);
339 safe_free(m_pszFormula);
340 m_pszFormula = pMsg->GetVariableStr(VID_DCI_FORMULA);
07a45e04
VK
341
342 // Update thresholds
343 dwNum = pMsg->GetVariableLong(VID_NUM_THRESHOLDS);
344 pNewThresholds = (DCI_THRESHOLD *)malloc(sizeof(DCI_THRESHOLD) * dwNum);
345 *ppdwMapIndex = (DWORD *)malloc(dwNum * sizeof(DWORD));
346 *ppdwMapId = (DWORD *)malloc(dwNum * sizeof(DWORD));
347 *pdwNumMaps = 0;
348
349 // Read all thresholds from message
350 for(i = 0, dwId = VID_DCI_THRESHOLD_BASE; i < dwNum; i++, dwId++)
351 {
352 pMsg->GetVariableBinary(dwId, (BYTE *)&pNewThresholds[i], sizeof(DCI_THRESHOLD));
353 pNewThresholds[i].dwId = ntohl(pNewThresholds[i].dwId);
354 }
355
06e7be2f
VK
356 // Check if some thresholds was deleted, and reposition others if needed
357 ppNewList = (Threshold **)malloc(sizeof(Threshold *) * dwNum);
07a45e04
VK
358 for(i = 0; i < m_dwNumThresholds; i++)
359 {
360 for(j = 0; j < dwNum; j++)
361 if (m_ppThresholdList[i]->Id() == pNewThresholds[j].dwId)
362 break;
363 if (j == dwNum)
364 {
365 // No threshold with that id in new list, delete it
366 delete m_ppThresholdList[i];
367 m_dwNumThresholds--;
368 memmove(&m_ppThresholdList[i], &m_ppThresholdList[i + 1], sizeof(Threshold *) * (m_dwNumThresholds - i));
369 i--;
370 }
06e7be2f
VK
371 else
372 {
373 // Move existing thresholds to appropriate positions in new list
374 ppNewList[j] = m_ppThresholdList[i];
375 }
07a45e04 376 }
06e7be2f
VK
377 safe_free(m_ppThresholdList);
378 m_ppThresholdList = ppNewList;
379 m_dwNumThresholds = dwNum;
07a45e04
VK
380
381 // Add or update thresholds
382 for(i = 0; i < dwNum; i++)
383 {
384 if (pNewThresholds[i].dwId == 0) // New threshold?
385 {
06e7be2f
VK
386 m_ppThresholdList[i] = new Threshold(this);
387 m_ppThresholdList[i]->CreateId();
07a45e04
VK
388
389 // Add index -> id mapping
390 (*ppdwMapIndex)[*pdwNumMaps] = i;
06e7be2f
VK
391 (*ppdwMapId)[*pdwNumMaps] = m_ppThresholdList[i]->Id();
392 (*pdwNumMaps)++;
07a45e04 393 }
06e7be2f 394 m_ppThresholdList[i]->UpdateFromMessage(&pNewThresholds[i]);
07a45e04
VK
395 }
396
397 safe_free(pNewThresholds);
398 Unlock();
9ed4eaff 399}
333ece94
VK
400
401
402//
403// Process new value
404//
405
406void DCItem::NewValue(DWORD dwTimeStamp, const char *pszOriginalValue)
407{
408 char szQuery[MAX_LINE_SIZE + 128];
409 ItemValue *pValue;
410
411 // Normally m_pNode shouldn't be NULL for polled items,
412 // but who knows...
413 if (m_pNode == NULL)
414 return;
415
416 // Create new ItemValue object and transform it as needed
417 pValue = new ItemValue(pszOriginalValue);
418 m_prevRawValue = *pValue;
419 Transform(*pValue);
420
421 // Save transformed value to database
422 sprintf(szQuery, "INSERT INTO idata_%ld (item_id,idata_timestamp,idata_value)"
423 " VALUES (%ld,%ld,'%s')", m_pNode->Id(), m_dwId, dwTimeStamp,
424 pValue->String());
425 QueueSQLRequest(szQuery);
426
427 // Check thresholds and add value to cache
428 CheckThresholds(*pValue);
429}
430
431
432//
433// Transform received value
434//
435
436void DCItem::Transform(ItemValue &value)
437{
438 switch(m_iDeltaCalculation)
439 {
440 case DCM_SIMPLE:
441 switch(m_iDataType)
442 {
443 case DCI_DT_INTEGER:
444 value = (long)value - (long)m_prevRawValue;
445 break;
446 case DCI_DT_INT64:
447 value = (INT64)value - (INT64)m_prevRawValue;
448 break;
449 case DCI_DT_FLOAT:
450 value = (double)value - (double)m_prevRawValue;
451 break;
452 default:
453 // Delta calculation is not supported for other types
454 break;
455 }
456 break;
457 case DCM_AVERAGE_PER_SECOND:
458 break;
459 case DCM_AVERAGE_PER_MINUTE:
460 break;
461 default: // Default is no transformation
462 break;
463 }
464}