implemented single housekeeping thread for all message wait queues
[public/netxms.git] / src / client / nxpush / nxpush.cpp
1 /*
2 ** nxpush - command line tool used to push DCI values to NetXMS server
3 ** Copyright (C) 2006-2015 Alex 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 **/
20
21 #include <nms_common.h>
22 #include <nms_agent.h>
23 #include <nms_util.h>
24 #include <nxclient.h>
25
26 #if HAVE_GETOPT_H
27 #include <getopt.h>
28 #endif
29
30 /**
31 * Session handle
32 */
33 static NXCSession *s_session = NULL;
34
35 /**
36 * Data to send
37 */
38 static ObjectArray<NXCPushData> s_data(16, 16, true);
39
40 /**
41 * options
42 */
43 static int s_optVerbose = 1;
44 static const char *s_optHost = NULL;
45 static const char *s_optUser = "guest";
46 static const char *s_optPassword = "";
47 static bool s_optEncrypt = false;
48 static int s_optBatchSize = 0;
49 static time_t s_timestamp = 0;
50
51 /**
52 * Split IDs, cleanup and add to the send s_data
53 */
54 static BOOL AddValuePair(char *name, char *value)
55 {
56 BOOL ret = TRUE;
57 UINT32 dciId = 0;
58 UINT32 nodeId = 0;
59 char *dciName = NULL;
60 char *nodeName = NULL;
61
62 nodeName = name;
63 dciName = strchr(name, ':');
64 if (dciName != NULL)
65 {
66 *dciName++ = 0;
67
68 nodeId = strtoul(nodeName, NULL, 10);
69 dciId = strtoul(dciName, NULL, 10);
70 }
71 else
72 {
73 ret = FALSE;
74 }
75
76 if (ret == TRUE)
77 {
78 if (s_optVerbose > 2)
79 {
80 _tprintf(_T("AddValuePair: dciID=\"%d\", nodeName=\"%hs\", dciName=\"%hs\", value=\"%hs\"\n"),
81 dciId, nodeName, dciName, value);
82 }
83
84 NXCPushData *p = new NXCPushData();
85 if (p != NULL)
86 {
87 p->dciId = dciId;
88 p->nodeId = nodeId;
89 if (dciId > 0)
90 {
91 p->dciName = NULL;
92 }
93 else
94 {
95 #ifdef UNICODE
96 p->dciName = WideStringFromMBString(dciName);
97 #else
98 p->dciName = strdup(dciName);
99 #endif
100 }
101
102 if (nodeId > 0)
103 {
104 p->nodeName = NULL;
105 }
106 else
107 {
108 #ifdef UNICODE
109 p->nodeName= WideStringFromMBString(nodeName);
110 #else
111 p->nodeName = strdup(nodeName);
112 #endif
113 }
114
115 #ifdef UNICODE
116 p->value = WideStringFromMBString(value);
117 #else
118 p->value = strdup(value);
119 #endif
120 s_data.add(p);
121 }
122 else
123 {
124 if (s_optVerbose > 0)
125 {
126 _tprintf(_T("new failed!: %s\n"), _tcserror(errno));
127 }
128 }
129 }
130
131 return ret;
132 }
133
134 /**
135 * Values parser - clean string and split by '='
136 */
137 static BOOL AddValue(char *pair)
138 {
139 BOOL ret = FALSE;
140 char *p = pair;
141 char *value = NULL;
142
143 if (s_optVerbose > 1)
144 _tprintf(_T("AddValue(): %hs\n"), pair);
145
146 for (p = pair; *p != 0; p++)
147 {
148 if (*p == '=' && value == NULL)
149 {
150 value = p;
151 }
152 if (*p == 0x0D || *p == 0x0A)
153 {
154 *p = 0;
155 break;
156 }
157 }
158
159 if (value != NULL)
160 {
161 *value++ = 0;
162
163 AddValuePair(pair, value);
164
165 ret = TRUE;
166 }
167
168 return ret;
169 }
170
171 /**
172 * Callback function for debug messages from client library
173 */
174 static void DebugCallback(const TCHAR *pMsg)
175 {
176 _tprintf(_T("NXCL: %s\n"), pMsg);
177 }
178
179 /**
180 * Initialize client library and connect to the server
181 */
182 static BOOL Startup()
183 {
184 BOOL ret = FALSE;
185 UINT32 dwResult;
186
187 #ifdef _WIN32
188 WSADATA wsaData;
189
190 if (WSAStartup(2, &wsaData) != 0)
191 {
192 if (s_optVerbose > 0)
193 {
194 _tprintf(_T("Unable to initialize Windows sockets\n"));
195 }
196 return FALSE;
197 }
198 #endif
199
200 if (!NXCInitialize())
201 {
202 if (s_optVerbose > 0)
203 {
204 _tprintf(_T("Failed to initialize NetXMS client library\n"));
205 }
206 }
207 else
208 {
209 if (s_optVerbose > 2)
210 {
211 NXCSetDebugCallback(DebugCallback);
212
213 _tprintf(_T("Connecting to \"%hs\" as \"%hs\"; encryption is %s\n"), s_optHost, s_optUser,
214 s_optEncrypt ? _T("enabled") : _T("disabled"));
215 }
216
217 #ifdef UNICODE
218 WCHAR *wHost = WideStringFromMBString(s_optHost);
219 WCHAR *wUser = WideStringFromMBString(s_optUser);
220 WCHAR *wPassword = WideStringFromMBString(s_optPassword);
221 #define _HOST wHost
222 #define _USER wUser
223 #define _PASSWD wPassword
224 #else
225 #define _HOST s_optHost
226 #define _USER s_optUser
227 #define _PASSWD s_optPassword
228 #endif
229
230 s_session = new NXCSession();
231 static UINT32 protocolVersions[] = { CPV_INDEX_PUSH };
232 dwResult = s_session->connect(_HOST, _USER, _PASSWD, s_optEncrypt ? NXCF_ENCRYPT : 0, _T("nxpush/") NETXMS_VERSION_STRING,
233 protocolVersions, sizeof(protocolVersions) / sizeof(UINT32));
234
235 #ifdef UNICODE
236 free(wHost);
237 free(wUser);
238 free(wPassword);
239 #endif
240 if (dwResult != RCC_SUCCESS)
241 {
242 if (s_optVerbose > 0)
243 {
244 _tprintf(_T("Unable to connect to the server: %s\n"), NXCGetErrorText(dwResult));
245 }
246 }
247 else
248 {
249 s_session->setCommandTimeout(5 * 1000);
250 ret = TRUE;
251 }
252 NXCShutdown();
253 }
254
255 return ret;
256 }
257
258 /**
259 * Send all DCIs
260 */
261 static BOOL Send()
262 {
263 BOOL ret = TRUE;
264 UINT32 errIdx;
265
266 int i, size;
267 int batches = 1;
268
269 if (s_optBatchSize == 0)
270 {
271 s_optBatchSize = s_data.size();
272 }
273
274 batches = s_data.size() / s_optBatchSize;
275 if (s_data.size() % s_optBatchSize != 0)
276 {
277 batches++;
278 }
279
280 for (i = 0; i < batches; i++)
281 {
282 size = min(s_optBatchSize, s_data.size() - (s_optBatchSize * i));
283
284 if (s_optVerbose > 1)
285 {
286 _tprintf(_T("Sending batch #%d with %d records\n"), i + 1, size);
287 }
288
289 if (s_optVerbose > 2)
290 {
291 for (int j = 0; j < size; j++)
292 {
293 NXCPushData *rec = s_data.get(s_optBatchSize * i + j);
294 _tprintf(_T("Record #%d: \"%s\" for %d(%s):%d(%s)\n"),
295 (s_optBatchSize * i) + j + 1,
296 rec->value,
297 rec->nodeId,
298 rec->nodeName != NULL ? rec->nodeName : _T("n/a"),
299 rec->dciId,
300 rec->dciName != NULL ? rec->dciName : _T("n/a"));
301 }
302 }
303
304 UINT32 dwResult = ((DataCollectionController *)(s_session->getController(CONTROLLER_DATA_COLLECTION)))->pushData(&s_data, s_timestamp, &errIdx);
305 if (dwResult != RCC_SUCCESS)
306 {
307 if (s_optVerbose > 0)
308 {
309 _tprintf(_T("Push failed at record #%d (#%d in batch): %s.\n"),
310 (i * s_optBatchSize) + errIdx + 1,
311 errIdx + 1,
312 NXCGetErrorText(dwResult));
313 }
314
315 ret = FALSE;
316 break;
317 }
318 else
319 {
320 if (s_optVerbose > 1)
321 {
322 _tprintf(_T("Done.\n"));
323 }
324 }
325 }
326
327 return ret;
328 }
329
330 /**
331 * Disconnect and cleanup
332 */
333 static BOOL Teardown()
334 {
335 if (s_session != NULL)
336 {
337 delete s_session;
338 }
339
340 for(int i = 0; i < s_data.size(); i++)
341 {
342 NXCPushData *d = s_data.get(i);
343 safe_free(d->dciName);
344 safe_free(d->nodeName);
345 safe_free(d->value);
346 }
347
348 return TRUE;
349 }
350
351 /**
352 * Command line options
353 */
354 #if HAVE_DECL_GETOPT_LONG
355 static struct option longOptions[] =
356 {
357 { (char *)"batchsize", required_argument, NULL, 'b' },
358 #ifndef _WIN32
359 { (char *)"codepage", required_argument, NULL, 'c' },
360 #endif
361 { (char *)"encrypt", no_argument, NULL, 'e' },
362 { (char *)"help", no_argument, NULL, 'h' },
363 { (char *)"host", required_argument, NULL, 'H' },
364 { (char *)"password", required_argument, NULL, 'P' },
365 { (char *)"quiet", no_argument, NULL, 'q' },
366 { (char *)"timestamp-unix", required_argument, NULL, 't' },
367 { (char *)"timestamp-text", required_argument, NULL, 'T' },
368 { (char *)"user", required_argument, NULL, 'u' },
369 { (char *)"verbose", no_argument, NULL, 'v' },
370 { (char *)"version", no_argument, NULL, 'V' },
371 { NULL, 0, NULL, 0 }
372 };
373 #endif
374
375 #ifdef _WIN32
376 #define SHORT_OPTIONS "b:ehH:P:qt:T:u:vV"
377 #else
378 #define SHORT_OPTIONS "b:c:ehH:P:qt:T:u:vV"
379 #endif
380
381 /**
382 * Print usage info
383 */
384 static void usage(char *argv0)
385 {
386 _tprintf(
387 _T("NetXMS PUSH Version ") NETXMS_VERSION_STRING _T("\n")
388 _T("Copyright (c) 2006-2015 Raden Solutions\n\n")
389 _T("Usage: %hs [OPTIONS] [server] [@batch_file] [values]\n")
390 _T(" \n")
391 _T("Options:\n")
392 #if HAVE_GETOPT_LONG
393 _T(" -b, --batchsize <size> Batch size (default is to send all data in one batch)\n")
394 #ifndef _WIN32
395 _T(" -c, --codepage <page> Codepage (default is %hs)\n")
396 #endif
397 _T(" -e, --encrypt Encrypt session.\n")
398 _T(" -h, --help Display this help message.\n")
399 _T(" -H, --host <host> Server address.\n")
400 _T(" -P, --password <password> Specify user's password. Default is empty.\n")
401 _T(" -q, --quiet Suppress all messages.\n")
402 _T(" -t, --timestamp-unix <time> Specify timestamp for data as UNIX timestamp.\n")
403 _T(" -T, --timestamp-text <time> Specify timestamp for data as YYYYMMDDhhmmss.\n")
404 _T(" -u, --user <user> Login to server as user. Default is \"guest\".\n")
405 _T(" -v, --verbose Enable verbose messages. Add twice for debug\n")
406 _T(" -V, --version Display version information.\n\n")
407 #else
408 _T(" -b <size> Batch size (default is to send all data in one batch)\n")
409 _T(" -c <page> Codepage (default is %hs)\n")
410 _T(" -e Encrypt session.\n")
411 _T(" -h Display this help message.\n")
412 _T(" -H <host> Server address.\n")
413 _T(" -P <password> Specify user's password. Default is empty.\n")
414 _T(" -q Suppress all messages.\n")
415 _T(" -t <time> Specify timestamp for data as UNIX timestamp.\n")
416 _T(" -T <time> Specify timestamp for data as YYYYMMDDhhmmss.\n")
417 _T(" -u <user> Login to server as user. Default is \"guest\".\n")
418 _T(" -v Enable verbose messages. Add twice for debug\n")
419 _T(" -V Display version information.\n\n")
420 #endif
421 _T("Notes:\n")
422 _T(" * Values should be given in the following format:\n")
423 _T(" node:dci=value\n")
424 _T(" where node and dci can be specified either by ID, object name, DNS name,\n")
425 _T(" or IP address. If you wish to specify node by DNS name or IP address,\n")
426 _T(" you should prefix it with @ character\n")
427 _T(" * First parameter will be used as \"host\" if -H/--host is unset\n")
428 _T(" * Name of batch file cannot contain character = (equality sign)\n")
429 _T("\n")
430 _T("Examples:\n")
431 _T(" Push two values to server 10.0.0.1 as user \"sender\" with password \"passwd\":\n")
432 _T(" nxpush -H 10.0.0.1 -u sender -P passwd 10:24=1 10:PushParam=4\n\n")
433 _T(" Push values from file to server 10.0.0.1 as user \"guest\" without password:\n")
434 _T(" nxpush 10.0.0.1 @file\n"), argv0
435 #ifndef _WIN32
436 , ICONV_DEFAULT_CODEPAGE
437 #endif
438 );
439 }
440
441 /**
442 * Entry point
443 */
444 int main(int argc, char *argv[])
445 {
446 int ret = 0;
447 int c;
448
449 InitThreadLibrary();
450
451 BOOL bStart = TRUE;
452 opterr = 0;
453 #if HAVE_DECL_GETOPT_LONG
454 while ((c = getopt_long(argc, argv, SHORT_OPTIONS, longOptions, NULL)) != -1)
455 #else
456 while ((c = getopt(argc, argv, SHORT_OPTIONS)) != -1)
457 #endif
458 {
459 switch(c)
460 {
461 case 'b': // batch size
462 s_optBatchSize = atoi(optarg); // 0 == unlimited
463 break;
464 #ifndef _WIN32
465 case 'c':
466 SetDefaultCodepage(optarg);
467 break;
468 #endif
469 case 'e': // user
470 s_optEncrypt = true;
471 break;
472 case 'h': // help
473 usage(argv[0]);
474 exit(1);
475 break;
476 case 'H': // host
477 s_optHost = optarg;
478 break;
479 case 'P': // password
480 s_optPassword = optarg;
481 break;
482 case 'q': // quiet
483 s_optVerbose = 0;
484 break;
485 case 't': // timestamp as UNIX time
486 s_timestamp = (time_t)strtoull(optarg, NULL, 0);
487 break;
488 case 'T': // timestamp as YYYYMMDDhhmmss
489 s_timestamp = ParseDateTimeA(optarg, 0);
490 break;
491 case 'u': // user
492 s_optUser = optarg;
493 break;
494 case 'v': // verbose
495 s_optVerbose++;
496 break;
497 case 'V': // version
498 _tprintf(_T("nxpush (") NETXMS_VERSION_STRING _T(")\n"));
499 exit(0);
500 break;
501 case '?':
502 exit(3);
503 break;
504 }
505 }
506
507 if (s_optHost == NULL && optind < argc)
508 {
509 s_optHost = argv[optind++];
510 }
511
512 if (optind == argc)
513 {
514 if (s_optVerbose > 0)
515 {
516 _tprintf(_T("Not enough arguments\n\n"));
517 #if HAVE_GETOPT_LONG
518 _tprintf(_T("Try nxpush --help for more information.\n"));
519 #else
520 _tprintf(_T("Try nxpush -h for more information.\n"));
521 #endif
522 }
523 exit(1);
524 }
525
526 // Parse
527 if (optind < argc)
528 {
529 while (optind < argc)
530 {
531 char *p = argv[optind];
532
533 if (((*p == '@') && (strchr(p, '=') == NULL)) || (*p == '-'))
534 {
535 FILE *fileHandle = stdin;
536
537 if (*p != '-')
538 {
539 fileHandle = fopen(p + 1, "r");
540 }
541
542 if (fileHandle != NULL)
543 {
544 char buffer[1024];
545
546 while (fgets(buffer, sizeof(buffer), fileHandle) != NULL)
547 {
548 AddValue(buffer);
549 }
550
551 if (fileHandle != stdin)
552 {
553 fclose(fileHandle);
554 }
555 }
556 else
557 {
558 if (s_optVerbose > 0)
559 {
560 _tprintf(_T("Cannot open \"%hs\": %s\n"), p + 1, _tcserror(errno));
561 }
562 }
563 }
564 else
565 {
566 AddValue(argv[optind]);
567 }
568
569 optind++;
570 }
571 }
572
573 if (s_data.size() > 0)
574 {
575 if (s_optVerbose > 0)
576 _tprintf(_T("%d values to send\n"), s_data.size());
577 if (Startup())
578 {
579 if (s_optVerbose > 0)
580 _tprintf(_T("Connected\n"));
581 if (Send())
582 {
583 if (s_optVerbose > 0)
584 _tprintf(_T("Data sent\n"));
585 }
586 else
587 {
588 if (s_optVerbose > 0)
589 _tprintf(_T("Data sending failed\n"));
590 ret = 3;
591 }
592 }
593 }
594 else
595 {
596 if (s_optVerbose > 0)
597 {
598 _tprintf(_T("No valid pairs found; nothing to send\n"));
599 ret = 2;
600 }
601 }
602 Teardown();
603
604 return ret;
605 }