default chart height on performance tab increased
[public/netxms.git] / src / server / core / job.cpp
CommitLineData
76b4edb5 1/*
ab621f39 2** NetXMS - Network Management System
509bb045 3** Copyright (C) 2003-2011 Victor Kirhenshtein
ab621f39
VK
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: job.cpp
20**
21**/
22
23#include "nxcore.h"
24
25
27de5dab
VK
26//
27// Externals
28//
29
967893bb 30void UnregisterJob(UINT32 jobId);
27de5dab
VK
31
32
ab621f39
VK
33//
34// Constructor
35//
36
967893bb 37ServerJob::ServerJob(const TCHAR *type, const TCHAR *description, UINT32 node, UINT32 userId, bool createOnHold)
ab621f39 38{
24dc5346 39 m_id = CreateUniqueId(IDG_JOB);
0edd0ab0 40 m_userId = userId;
ab621f39
VK
41 m_type = _tcsdup(CHECK_NULL(type));
42 m_description = _tcsdup(CHECK_NULL(description));
509bb045 43 m_status = createOnHold ? JOB_ON_HOLD : JOB_PENDING;
3929b1ca
VK
44 m_lastStatusChange = time(NULL);
45 m_autoCancelDelay = 0;
ab621f39 46 m_remoteNode = node;
8134d3a3 47 m_resolvedObject = FindObjectById(m_remoteNode);
ab621f39
VK
48 m_progress = 0;
49 m_failureMessage = NULL;
50 m_owningQueue = NULL;
51 m_workerThread = INVALID_THREAD_HANDLE;
3929b1ca
VK
52 m_lastNotification = 0;
53 m_notificationLock = MutexCreate();
8134d3a3 54 m_blockNextJobsOnFailure = false;
878b4261 55 createHistoryRecord();
ab621f39
VK
56}
57
58
59//
60// Destructor
61//
62
63ServerJob::~ServerJob()
64{
27de5dab
VK
65 UnregisterJob(m_id);
66
ab621f39
VK
67 ThreadJoin(m_workerThread);
68
69 safe_free(m_type);
70 safe_free(m_description);
712dd47d 71 safe_free(m_failureMessage);
3929b1ca
VK
72 MutexDestroy(m_notificationLock);
73}
74
75
76//
77// Send notification to clients
78//
79
80void ServerJob::sendNotification(ClientSession *session, void *arg)
81{
82 ServerJob *job = (ServerJob *)arg;
6b8e9f96 83 if (job->m_resolvedObject->checkAccessRights(session->getUserId(), OBJECT_ACCESS_READ))
d3a7cf4c 84 session->postMessage(&job->m_notificationMessage);
3929b1ca
VK
85}
86
87
88//
89// Notify clients
90//
91
92void ServerJob::notifyClients(bool isStatusChange)
93{
8134d3a3
VK
94 if (m_resolvedObject == NULL)
95 return;
96
3929b1ca 97 time_t t = time(NULL);
901a5a9b
VK
98 if (!isStatusChange && (t - m_lastNotification < 3))
99 return; // Don't send progress notifications often then every 3 seconds
100 m_lastNotification = t;
3929b1ca 101
c17f6cbc 102 MutexLock(m_notificationLock);
3929b1ca
VK
103 m_notificationMessage.SetCode(CMD_JOB_CHANGE_NOTIFICATION);
104 fillMessage(&m_notificationMessage);
8134d3a3 105 EnumerateClientSessions(ServerJob::sendNotification, this);
3929b1ca
VK
106 MutexUnlock(m_notificationLock);
107}
108
109
110//
111// Change status
112//
113
114void ServerJob::changeStatus(ServerJobStatus newStatus)
115{
116 m_status = newStatus;
117 m_lastStatusChange = time(NULL);
118 notifyClients(true);
ab621f39
VK
119}
120
121
122//
123// Set owning queue
124//
125
126void ServerJob::setOwningQueue(ServerJobQueue *queue)
127{
128 m_owningQueue = queue;
3929b1ca 129 notifyClients(true);
ab621f39
VK
130}
131
132
76b4edb5 133/**
134 * Update progress
135 */
ab621f39
VK
136
137void ServerJob::markProgress(int pctCompleted)
138{
139 if ((pctCompleted > m_progress) && (pctCompleted <= 100))
3929b1ca 140 {
ab621f39 141 m_progress = pctCompleted;
3929b1ca
VK
142 notifyClients(false);
143 }
ab621f39
VK
144}
145
146
147//
148// Worker thread starter
149//
150
151THREAD_RESULT THREAD_CALL ServerJob::WorkerThreadStarter(void *arg)
152{
129a1ce0
VK
153 ServerJob *job = (ServerJob *)arg;
154 DbgPrintf(4, _T("Job %d started"), job->m_id);
878b4261 155 job->updateHistoryRecord(true);
ab621f39 156
129a1ce0
VK
157 if (job->run())
158 {
3929b1ca 159 job->changeStatus(JOB_COMPLETED);
129a1ce0 160 }
ab621f39 161 else
129a1ce0 162 {
878b4261
VK
163 if (job->m_status == JOB_CANCEL_PENDING)
164 job->changeStatus(JOB_CANCELLED);
165 else
901a5a9b 166 job->changeStatus(JOB_FAILED);
129a1ce0
VK
167 }
168 job->m_workerThread = INVALID_THREAD_HANDLE;
ab621f39 169
878b4261
VK
170 DbgPrintf(4, _T("Job %d finished, status=%s"), job->m_id, (job->m_status == JOB_COMPLETED) ? _T("COMPLETED") : ((job->m_status == JOB_CANCELLED) ? _T("CANCELLED") : _T("FAILED")));
171 job->updateHistoryRecord(false);
ab621f39 172
129a1ce0
VK
173 if (job->m_owningQueue != NULL)
174 job->m_owningQueue->jobCompleted(job);
ab621f39
VK
175 return THREAD_OK;
176}
177
178
179//
180// Start job
181//
182
183void ServerJob::start()
184{
185 m_status = JOB_ACTIVE;
186 m_workerThread = ThreadCreateEx(WorkerThreadStarter, 0, this);
187}
188
189
190//
191// Cancel job
192//
193
194bool ServerJob::cancel()
195{
f40831eb
VK
196 switch(m_status)
197 {
198 case JOB_COMPLETED:
f2665675 199 case JOB_CANCEL_PENDING:
f40831eb
VK
200 return false;
201 case JOB_ACTIVE:
f2665675
VK
202 if (!onCancel())
203 return false;
204 changeStatus(JOB_CANCEL_PENDING);
205 return true;
f40831eb 206 default:
3929b1ca 207 changeStatus(JOB_CANCELLED);
f40831eb
VK
208 return true;
209 }
ab621f39
VK
210}
211
212
509bb045
VK
213//
214// Hold job
215//
216
217bool ServerJob::hold()
218{
219 if (m_status == JOB_PENDING)
220 {
221 changeStatus(JOB_ON_HOLD);
222 return true;
223 }
224 return false;
225}
226
227
228//
229// Unhold job
230//
231
232bool ServerJob::unhold()
233{
234 if (m_status == JOB_ON_HOLD)
235 {
236 changeStatus(JOB_PENDING);
237 return true;
238 }
239 return false;
240}
241
242
ab621f39
VK
243//
244// Default run (empty)
245//
246
247bool ServerJob::run()
248{
249 return true;
250}
251
252
253//
254// Default cancel handler
255//
256
257bool ServerJob::onCancel()
258{
259 return false;
260}
261
262
263//
264// Set failure message
265//
266
267void ServerJob::setFailureMessage(const TCHAR *msg)
268{
269 safe_free(m_failureMessage);
270 m_failureMessage = (msg != NULL) ? _tcsdup(msg) : NULL;
271}
3929b1ca
VK
272
273
274//
275// Set description
276//
277
278void ServerJob::setDescription(const TCHAR *description)
76b4edb5 279{
3929b1ca 280 safe_free(m_description);
76b4edb5 281 m_description = _tcsdup(description);
3929b1ca
VK
282}
283
284
285//
286// Fill NXCP message with job's data
287//
288
289void ServerJob::fillMessage(CSCPMessage *msg)
290{
291 msg->SetVariable(VID_JOB_ID, m_id);
0edd0ab0 292 msg->SetVariable(VID_USER_ID, m_userId);
3929b1ca
VK
293 msg->SetVariable(VID_JOB_TYPE, m_type);
294 msg->SetVariable(VID_OBJECT_ID, m_remoteNode);
295 msg->SetVariable(VID_DESCRIPTION, CHECK_NULL_EX(m_description));
296 msg->SetVariable(VID_JOB_STATUS, (WORD)m_status);
297 msg->SetVariable(VID_JOB_PROGRESS, (WORD)m_progress);
8134d3a3 298 if (m_status == JOB_FAILED)
509bb045 299 msg->SetVariable(VID_FAILURE_MESSAGE, (m_failureMessage != NULL) ? m_failureMessage : _T("Internal error"));
8134d3a3
VK
300 else
301 msg->SetVariable(VID_FAILURE_MESSAGE, CHECK_NULL_EX(m_failureMessage));
3929b1ca 302}
878b4261 303
77c78cbf
VK
304/**
305 * Create record in job history table
306 */
878b4261
VK
307void ServerJob::createHistoryRecord()
308{
309 DB_HANDLE hdb = DBConnectionPoolAcquireConnection();
76b4edb5 310
311 DB_STATEMENT hStmt = DBPrepare(hdb,
878b4261
VK
312 _T("INSERT INTO job_history (id,time_created,time_started,time_finished,job_type,")
313 _T("description,node_id,user_id,status) VALUES (?,?,0,0,?,?,?,?,?)"));
314 if (hStmt != NULL)
315 {
316 DBBind(hStmt, 1, DB_SQLTYPE_INTEGER, m_id);
967893bb 317 DBBind(hStmt, 2, DB_SQLTYPE_INTEGER, (UINT32)time(NULL));
878b4261
VK
318 DBBind(hStmt, 3, DB_SQLTYPE_VARCHAR, m_type, DB_BIND_STATIC);
319 DBBind(hStmt, 4, DB_SQLTYPE_VARCHAR, CHECK_NULL_EX(m_description), DB_BIND_STATIC);
320 DBBind(hStmt, 5, DB_SQLTYPE_INTEGER, m_remoteNode);
321 DBBind(hStmt, 6, DB_SQLTYPE_INTEGER, m_userId);
322 DBBind(hStmt, 7, DB_SQLTYPE_INTEGER, (LONG)m_status);
323 DBExecute(hStmt);
324 DBFreeStatement(hStmt);
325 }
326 DBConnectionPoolReleaseConnection(hdb);
327}
328
77c78cbf
VK
329/**
330 * Update job history record
331 */
878b4261
VK
332void ServerJob::updateHistoryRecord(bool onStart)
333{
334 DB_HANDLE hdb = DBConnectionPoolAcquireConnection();
76b4edb5 335
336 DB_STATEMENT hStmt = DBPrepare(hdb,
337 onStart ?
338 _T("UPDATE job_history SET time_started=?,status=?,description=?,additional_info=? WHERE id=?") :
24dc5346 339 _T("UPDATE job_history SET time_finished=?,status=?,description=?,additional_info=?,failure_message=? WHERE id=?"));
76b4edb5 340
878b4261
VK
341 if (hStmt != NULL)
342 {
967893bb 343 DBBind(hStmt, 1, DB_SQLTYPE_INTEGER, (UINT32)time(NULL));
878b4261 344 DBBind(hStmt, 2, DB_SQLTYPE_INTEGER, (LONG)m_status);
24dc5346
VK
345 DBBind(hStmt, 3, DB_SQLTYPE_VARCHAR, CHECK_NULL_EX(m_description), DB_BIND_STATIC);
346 DBBind(hStmt, 4, DB_SQLTYPE_VARCHAR, getAdditionalInfo(), DB_BIND_TRANSIENT);
878b4261
VK
347 if (onStart)
348 {
24dc5346 349 DBBind(hStmt, 5, DB_SQLTYPE_INTEGER, m_id);
878b4261
VK
350 }
351 else
352 {
24dc5346
VK
353 DBBind(hStmt, 5, DB_SQLTYPE_VARCHAR, CHECK_NULL_EX(m_failureMessage), DB_BIND_STATIC);
354 DBBind(hStmt, 6, DB_SQLTYPE_INTEGER, m_id);
878b4261
VK
355 }
356 DBExecute(hStmt);
357 DBFreeStatement(hStmt);
358 }
359 DBConnectionPoolReleaseConnection(hdb);
360}
24dc5346 361
77c78cbf
VK
362/**
363 * Get additional info for logging
364 */
24dc5346
VK
365const TCHAR *ServerJob::getAdditionalInfo()
366{
367 return _T("");
368}