pytdb: Add support for tdb_enable_seqnum, tdb_get_seqnum and tdb_increment_seqnum_non...
[metze/ctdb/wip.git] / server / ctdb_control.c
1 /* 
2    ctdb_control protocol code
3
4    Copyright (C) Andrew Tridgell  2007
5
6    This program is free software; you can redistribute it and/or modify
7    it under the terms of the GNU General Public License as published by
8    the Free Software Foundation; either version 3 of the License, or
9    (at your option) any later version.
10    
11    This program is distributed in the hope that it will be useful,
12    but WITHOUT ANY WARRANTY; without even the implied warranty of
13    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
14    GNU General Public License for more details.
15    
16    You should have received a copy of the GNU General Public License
17    along with this program; if not, see <http://www.gnu.org/licenses/>.
18 */
19 #include "includes.h"
20 #include "lib/tevent/tevent.h"
21 #include "lib/tdb/include/tdb.h"
22 #include "system/network.h"
23 #include "system/filesys.h"
24 #include "system/wait.h"
25 #include "../include/ctdb_private.h"
26 #include "lib/util/dlinklist.h"
27 #include "db_wrap.h"
28
29
30 struct ctdb_control_state {
31         struct ctdb_context *ctdb;
32         uint32_t reqid;
33         ctdb_control_callback_fn_t callback;
34         void *private_data;
35         unsigned flags;
36 };
37
38
39 /*
40   dump talloc memory hierarchy, returning it as a blob to the client
41  */
42 int32_t ctdb_dump_memory(struct ctdb_context *ctdb, TDB_DATA *outdata)
43 {
44         /* dump to a file, then send the file as a blob */
45         FILE *f;
46         long fsize;
47         f = tmpfile();
48         if (f == NULL) {
49                 DEBUG(DEBUG_ERR,(__location__ " Unable to open tmpfile - %s\n", strerror(errno)));
50                 return -1;
51         }
52         talloc_report_full(NULL, f);
53         fsize = ftell(f);
54         rewind(f);
55         outdata->dptr = talloc_size(outdata, fsize);
56         CTDB_NO_MEMORY(ctdb, outdata->dptr);
57         outdata->dsize = fread(outdata->dptr, 1, fsize, f);
58         fclose(f);
59         if (outdata->dsize != fsize) {
60                 DEBUG(DEBUG_ERR,(__location__ " Unable to read tmpfile\n"));
61                 return -1;
62         }
63         return 0;
64 }
65
66
67 /*
68   process a control request
69  */
70 static int32_t ctdb_control_dispatch(struct ctdb_context *ctdb, 
71                                      struct ctdb_req_control *c,
72                                      TDB_DATA indata,
73                                      TDB_DATA *outdata, uint32_t srcnode,
74                                      const char **errormsg,
75                                      bool *async_reply)
76 {
77         uint32_t opcode = c->opcode;
78         uint64_t srvid = c->srvid;
79         uint32_t client_id = c->client_id;
80
81         switch (opcode) {
82         case CTDB_CONTROL_PROCESS_EXISTS: {
83                 CHECK_CONTROL_DATA_SIZE(sizeof(pid_t));
84                 return ctdb_control_process_exists(ctdb, *(pid_t *)indata.dptr);
85         }
86
87         case CTDB_CONTROL_SET_DEBUG: {
88                 CHECK_CONTROL_DATA_SIZE(sizeof(int32_t));
89                 LogLevel = *(int32_t *)indata.dptr;
90                 return 0;
91         }
92
93         case CTDB_CONTROL_GET_DEBUG: {
94                 CHECK_CONTROL_DATA_SIZE(0);
95                 outdata->dptr = (uint8_t *)&LogLevel;
96                 outdata->dsize = sizeof(LogLevel);
97                 return 0;
98         }
99
100         case CTDB_CONTROL_STATISTICS: {
101                 int i;
102                 CHECK_CONTROL_DATA_SIZE(0);
103                 ctdb->statistics.memory_used = talloc_total_size(NULL);
104                 ctdb->statistics.frozen = 0;
105                 for (i=1; i<= NUM_DB_PRIORITIES; i++) {
106                         if (ctdb->freeze_mode[i] == CTDB_FREEZE_FROZEN) {
107                                 ctdb->statistics.frozen = 1;
108                         }
109                 }
110                 ctdb->statistics.recovering = (ctdb->recovery_mode == CTDB_RECOVERY_ACTIVE);
111                 ctdb->statistics.statistics_current_time = timeval_current();
112
113                 outdata->dptr = (uint8_t *)&ctdb->statistics;
114                 outdata->dsize = sizeof(ctdb->statistics);
115                 return 0;
116         }
117
118         case CTDB_CONTROL_GET_ALL_TUNABLES: {
119                 CHECK_CONTROL_DATA_SIZE(0);
120                 outdata->dptr = (uint8_t *)&ctdb->tunable;
121                 outdata->dsize = sizeof(ctdb->tunable);
122                 return 0;
123         }
124
125         case CTDB_CONTROL_DUMP_MEMORY: {
126                 CHECK_CONTROL_DATA_SIZE(0);
127                 return ctdb_dump_memory(ctdb, outdata);
128         }
129
130         case CTDB_CONTROL_STATISTICS_RESET: {
131                 CHECK_CONTROL_DATA_SIZE(0);
132                 ZERO_STRUCT(ctdb->statistics);
133                 ctdb->statistics.statistics_start_time = timeval_current();
134                 return 0;
135         }
136
137         case CTDB_CONTROL_GETVNNMAP:
138                 return ctdb_control_getvnnmap(ctdb, opcode, indata, outdata);
139
140         case CTDB_CONTROL_GET_DBMAP:
141                 return ctdb_control_getdbmap(ctdb, opcode, indata, outdata);
142
143         case CTDB_CONTROL_GET_NODEMAPv4:
144                 return ctdb_control_getnodemapv4(ctdb, opcode, indata, outdata);
145
146         case CTDB_CONTROL_GET_NODEMAP:
147                 return ctdb_control_getnodemap(ctdb, opcode, indata, outdata);
148
149         case CTDB_CONTROL_RELOAD_NODES_FILE:
150                 CHECK_CONTROL_DATA_SIZE(0);
151                 return ctdb_control_reload_nodes_file(ctdb, opcode);
152
153         case CTDB_CONTROL_SETVNNMAP:
154                 return ctdb_control_setvnnmap(ctdb, opcode, indata, outdata);
155
156         case CTDB_CONTROL_PULL_DB: 
157                 CHECK_CONTROL_DATA_SIZE(sizeof(struct ctdb_control_pulldb));
158                 return ctdb_control_pull_db(ctdb, indata, outdata);
159
160         case CTDB_CONTROL_SET_DMASTER: 
161                 CHECK_CONTROL_DATA_SIZE(sizeof(struct ctdb_control_set_dmaster));
162                 return ctdb_control_set_dmaster(ctdb, indata);
163
164         case CTDB_CONTROL_PUSH_DB:
165                 return ctdb_control_push_db(ctdb, indata);
166
167         case CTDB_CONTROL_GET_RECMODE: {
168                 int i;
169                 if (ctdb->recovery_mode == CTDB_RECOVERY_ACTIVE) {
170                         return CTDB_RECOVERY_ACTIVE;
171                 }                 
172                 for (i=1; i<=NUM_DB_PRIORITIES; i++) {
173                         if (ctdb->freeze_mode[i] == CTDB_FREEZE_FROZEN) {
174                                 return CTDB_RECOVERY_ACTIVE;
175                         }
176                 }
177                 return CTDB_RECOVERY_NORMAL;
178         }
179
180         case CTDB_CONTROL_SET_RECMASTER: {
181                 return ctdb_control_set_recmaster(ctdb, opcode, indata);
182         }
183
184         case CTDB_CONTROL_GET_RECMASTER:
185                 return ctdb->recovery_master;
186
187         case CTDB_CONTROL_GET_PID:
188                 return getpid();
189
190         case CTDB_CONTROL_GET_PNN:
191                 return ctdb->pnn;
192
193         case CTDB_CONTROL_PING:
194                 CHECK_CONTROL_DATA_SIZE(0);
195                 return ctdb->statistics.num_clients;
196
197         case CTDB_CONTROL_GET_DBNAME: {
198                 uint32_t db_id;
199                 struct ctdb_db_context *ctdb_db;
200
201                 CHECK_CONTROL_DATA_SIZE(sizeof(db_id));
202                 db_id = *(uint32_t *)indata.dptr;
203                 ctdb_db = find_ctdb_db(ctdb, db_id);
204                 if (ctdb_db == NULL) return -1;
205                 outdata->dptr = discard_const(ctdb_db->db_name);
206                 outdata->dsize = strlen(ctdb_db->db_name)+1;
207                 return 0;
208         }
209
210         case CTDB_CONTROL_GETDBPATH: {
211                 uint32_t db_id;
212                 struct ctdb_db_context *ctdb_db;
213
214                 CHECK_CONTROL_DATA_SIZE(sizeof(db_id));
215                 db_id = *(uint32_t *)indata.dptr;
216                 ctdb_db = find_ctdb_db(ctdb, db_id);
217                 if (ctdb_db == NULL) return -1;
218                 outdata->dptr = discard_const(ctdb_db->db_path);
219                 outdata->dsize = strlen(ctdb_db->db_path)+1;
220                 return 0;
221         }
222
223         case CTDB_CONTROL_DB_ATTACH:
224                 return ctdb_control_db_attach(ctdb, indata, outdata, srvid, false);
225
226         case CTDB_CONTROL_DB_ATTACH_PERSISTENT:
227                 return ctdb_control_db_attach(ctdb, indata, outdata, srvid, true);
228
229         case CTDB_CONTROL_SET_CALL: {
230                 struct ctdb_control_set_call *sc = 
231                         (struct ctdb_control_set_call *)indata.dptr;
232                 CHECK_CONTROL_DATA_SIZE(sizeof(struct ctdb_control_set_call));
233                 return ctdb_daemon_set_call(ctdb, sc->db_id, sc->fn, sc->id);
234         }
235
236         case CTDB_CONTROL_TRAVERSE_START:
237                 CHECK_CONTROL_DATA_SIZE(sizeof(struct ctdb_traverse_start));
238                 return ctdb_control_traverse_start(ctdb, indata, outdata, srcnode, client_id);
239
240         case CTDB_CONTROL_TRAVERSE_ALL:
241                 return ctdb_control_traverse_all(ctdb, indata, outdata);
242
243         case CTDB_CONTROL_TRAVERSE_DATA:
244                 return ctdb_control_traverse_data(ctdb, indata, outdata);
245
246         case CTDB_CONTROL_TRAVERSE_KILL:
247                 CHECK_CONTROL_DATA_SIZE(sizeof(struct ctdb_traverse_start));
248                 return ctdb_control_traverse_kill(ctdb, indata, outdata, srcnode);
249
250         case CTDB_CONTROL_REGISTER_SRVID:
251                 return daemon_register_message_handler(ctdb, client_id, srvid);
252
253         case CTDB_CONTROL_DEREGISTER_SRVID:
254                 return daemon_deregister_message_handler(ctdb, client_id, srvid);
255
256         case CTDB_CONTROL_ENABLE_SEQNUM:
257                 CHECK_CONTROL_DATA_SIZE(sizeof(uint32_t));
258                 return ctdb_ltdb_enable_seqnum(ctdb, *(uint32_t *)indata.dptr);
259
260         case CTDB_CONTROL_UPDATE_SEQNUM:
261                 CHECK_CONTROL_DATA_SIZE(sizeof(uint32_t));              
262                 return ctdb_ltdb_update_seqnum(ctdb, *(uint32_t *)indata.dptr, srcnode);
263
264         case CTDB_CONTROL_FREEZE:
265                 CHECK_CONTROL_DATA_SIZE(0);
266                 return ctdb_control_freeze(ctdb, c, async_reply);
267
268         case CTDB_CONTROL_THAW:
269                 CHECK_CONTROL_DATA_SIZE(0);
270                 return ctdb_control_thaw(ctdb, (uint32_t)c->srvid);
271
272         case CTDB_CONTROL_SET_RECMODE:
273                 CHECK_CONTROL_DATA_SIZE(sizeof(uint32_t));              
274                 return ctdb_control_set_recmode(ctdb, c, indata, async_reply, errormsg);
275
276         case CTDB_CONTROL_GET_MONMODE: 
277                 CHECK_CONTROL_DATA_SIZE(0);
278                 return ctdb_monitoring_mode(ctdb);
279                 
280         case CTDB_CONTROL_ENABLE_MONITOR: 
281                 CHECK_CONTROL_DATA_SIZE(0);
282                 ctdb_enable_monitoring(ctdb);
283                 return 0;
284         
285         case CTDB_CONTROL_RUN_EVENTSCRIPTS: 
286                 return ctdb_run_eventscripts(ctdb, c, indata, async_reply);
287
288         case CTDB_CONTROL_DISABLE_MONITOR: 
289                 CHECK_CONTROL_DATA_SIZE(0);
290                 ctdb_disable_monitoring(ctdb);
291                 return 0;
292
293         case CTDB_CONTROL_SHUTDOWN:
294                 ctdb_stop_recoverd(ctdb);
295                 ctdb_stop_keepalive(ctdb);
296                 ctdb_stop_monitoring(ctdb);
297                 ctdb_release_all_ips(ctdb);
298                 if (ctdb->methods != NULL) {
299                         ctdb->methods->shutdown(ctdb);
300                 }
301                 ctdb_event_script(ctdb, CTDB_EVENT_SHUTDOWN);
302                 DEBUG(DEBUG_NOTICE,("Received SHUTDOWN command. Stopping CTDB daemon.\n"));
303                 exit(0);
304
305         case CTDB_CONTROL_TAKEOVER_IPv4:
306                 CHECK_CONTROL_DATA_SIZE(sizeof(struct ctdb_public_ipv4));
307                 return ctdb_control_takeover_ipv4(ctdb, c, indata, async_reply);
308
309         case CTDB_CONTROL_TAKEOVER_IP:
310                 CHECK_CONTROL_DATA_SIZE(sizeof(struct ctdb_public_ip));
311                 return ctdb_control_takeover_ip(ctdb, c, indata, async_reply);
312
313         case CTDB_CONTROL_RELEASE_IPv4:
314                 CHECK_CONTROL_DATA_SIZE(sizeof(struct ctdb_public_ipv4));
315                 return ctdb_control_release_ipv4(ctdb, c, indata, async_reply);
316
317         case CTDB_CONTROL_RELEASE_IP:
318                 CHECK_CONTROL_DATA_SIZE(sizeof(struct ctdb_public_ip));
319                 return ctdb_control_release_ip(ctdb, c, indata, async_reply);
320
321         case CTDB_CONTROL_GET_PUBLIC_IPSv4:
322                 CHECK_CONTROL_DATA_SIZE(0);
323                 return ctdb_control_get_public_ipsv4(ctdb, c, outdata);
324
325         case CTDB_CONTROL_GET_PUBLIC_IPS:
326                 CHECK_CONTROL_DATA_SIZE(0);
327                 return ctdb_control_get_public_ips(ctdb, c, outdata);
328
329         case CTDB_CONTROL_TCP_CLIENT: 
330                 return ctdb_control_tcp_client(ctdb, client_id, indata);
331
332         case CTDB_CONTROL_STARTUP: 
333                 CHECK_CONTROL_DATA_SIZE(0);
334                 return ctdb_control_startup(ctdb, srcnode);
335
336         case CTDB_CONTROL_TCP_ADD: 
337                 CHECK_CONTROL_DATA_SIZE(sizeof(struct ctdb_tcp_connection));
338                 return ctdb_control_tcp_add(ctdb, indata, false);
339
340         case CTDB_CONTROL_TCP_ADD_DELAYED_UPDATE: 
341                 CHECK_CONTROL_DATA_SIZE(sizeof(struct ctdb_tcp_connection));
342                 return ctdb_control_tcp_add(ctdb, indata, true);
343
344         case CTDB_CONTROL_TCP_REMOVE: 
345                 CHECK_CONTROL_DATA_SIZE(sizeof(struct ctdb_tcp_connection));
346                 return ctdb_control_tcp_remove(ctdb, indata);
347
348         case CTDB_CONTROL_SET_TUNABLE:
349                 return ctdb_control_set_tunable(ctdb, indata);
350
351         case CTDB_CONTROL_GET_TUNABLE:
352                 return ctdb_control_get_tunable(ctdb, indata, outdata);
353
354         case CTDB_CONTROL_LIST_TUNABLES:
355                 return ctdb_control_list_tunables(ctdb, outdata);
356
357         case CTDB_CONTROL_MODIFY_FLAGS:
358                 CHECK_CONTROL_DATA_SIZE(sizeof(struct ctdb_node_flag_change));
359                 return ctdb_control_modflags(ctdb, indata);
360
361         case CTDB_CONTROL_KILL_TCP: 
362                 CHECK_CONTROL_DATA_SIZE(sizeof(struct ctdb_control_killtcp));
363                 return ctdb_control_kill_tcp(ctdb, indata);
364
365         case CTDB_CONTROL_GET_TCP_TICKLE_LIST:
366                 CHECK_CONTROL_DATA_SIZE(sizeof(ctdb_sock_addr));
367                 return ctdb_control_get_tcp_tickle_list(ctdb, indata, outdata);
368
369         case CTDB_CONTROL_SET_TCP_TICKLE_LIST:
370                 /* data size is verified in the called function */
371                 return ctdb_control_set_tcp_tickle_list(ctdb, indata);
372
373         case CTDB_CONTROL_REGISTER_SERVER_ID: 
374                 CHECK_CONTROL_DATA_SIZE(sizeof(struct ctdb_server_id));
375                 return ctdb_control_register_server_id(ctdb, client_id, indata);
376
377         case CTDB_CONTROL_UNREGISTER_SERVER_ID: 
378                 CHECK_CONTROL_DATA_SIZE(sizeof(struct ctdb_server_id));
379                 return ctdb_control_unregister_server_id(ctdb, indata);
380
381         case CTDB_CONTROL_CHECK_SERVER_ID: 
382                 CHECK_CONTROL_DATA_SIZE(sizeof(struct ctdb_server_id));
383                 return ctdb_control_check_server_id(ctdb, indata);
384
385         case CTDB_CONTROL_GET_SERVER_ID_LIST:
386                 CHECK_CONTROL_DATA_SIZE(0);
387                 return ctdb_control_get_server_id_list(ctdb, outdata);
388
389         case CTDB_CONTROL_PERSISTENT_STORE:
390                 return ctdb_control_persistent_store(ctdb, c, indata, async_reply);
391
392         case CTDB_CONTROL_UPDATE_RECORD:
393                 return ctdb_control_update_record(ctdb, c, indata, async_reply);
394
395         case CTDB_CONTROL_SEND_GRATIOUS_ARP:
396                 return ctdb_control_send_gratious_arp(ctdb, indata);
397
398         case CTDB_CONTROL_TRANSACTION_START:
399                 CHECK_CONTROL_DATA_SIZE(sizeof(uint32_t));
400                 return ctdb_control_transaction_start(ctdb, *(uint32_t *)indata.dptr);
401
402         case CTDB_CONTROL_TRANSACTION_COMMIT:
403                 CHECK_CONTROL_DATA_SIZE(sizeof(uint32_t));
404                 return ctdb_control_transaction_commit(ctdb, *(uint32_t *)indata.dptr);
405
406         case CTDB_CONTROL_WIPE_DATABASE:
407                 CHECK_CONTROL_DATA_SIZE(sizeof(struct ctdb_control_wipe_database));
408                 return ctdb_control_wipe_database(ctdb, indata);
409
410         case CTDB_CONTROL_UPTIME:
411                 return ctdb_control_uptime(ctdb, outdata);
412
413         case CTDB_CONTROL_START_RECOVERY:
414                 return ctdb_control_start_recovery(ctdb, c, async_reply);
415
416         case CTDB_CONTROL_END_RECOVERY:
417                 return ctdb_control_end_recovery(ctdb, c, async_reply);
418
419         case CTDB_CONTROL_TRY_DELETE_RECORDS:
420                 return ctdb_control_try_delete_records(ctdb, indata, outdata);
421
422         case CTDB_CONTROL_ADD_PUBLIC_IP:
423                 return ctdb_control_add_public_address(ctdb, indata);
424
425         case CTDB_CONTROL_DEL_PUBLIC_IP:
426                 return ctdb_control_del_public_address(ctdb, indata);
427
428         case CTDB_CONTROL_GET_CAPABILITIES:
429                 return ctdb_control_get_capabilities(ctdb, outdata);
430
431         case CTDB_CONTROL_START_PERSISTENT_UPDATE:
432                 return ctdb_control_start_persistent_update(ctdb, c, indata);
433
434         case CTDB_CONTROL_CANCEL_PERSISTENT_UPDATE:
435                 return ctdb_control_cancel_persistent_update(ctdb, c, indata);
436
437         case CTDB_CONTROL_TRANS2_COMMIT:
438         case CTDB_CONTROL_TRANS2_COMMIT_RETRY:
439                 return ctdb_control_trans2_commit(ctdb, c, indata, async_reply);
440
441         case CTDB_CONTROL_TRANS2_ERROR:
442                 return ctdb_control_trans2_error(ctdb, c);
443
444         case CTDB_CONTROL_TRANS2_FINISHED:
445                 return ctdb_control_trans2_finished(ctdb, c);
446
447         case CTDB_CONTROL_TRANS2_ACTIVE:
448                 CHECK_CONTROL_DATA_SIZE(sizeof(uint32_t));
449                 return ctdb_control_trans2_active(ctdb, c, *(uint32_t *)indata.dptr);
450
451         case CTDB_CONTROL_TRANS3_COMMIT:
452                 return ctdb_control_trans3_commit(ctdb, c, indata, async_reply);
453
454         case CTDB_CONTROL_RECD_PING:
455                 CHECK_CONTROL_DATA_SIZE(0);
456                 return ctdb_control_recd_ping(ctdb);
457
458         case CTDB_CONTROL_GET_EVENT_SCRIPT_STATUS:
459                 CHECK_CONTROL_DATA_SIZE(sizeof(uint32_t));
460                 return ctdb_control_get_event_script_status(ctdb, *(uint32_t *)indata.dptr, outdata);
461
462         case CTDB_CONTROL_RECD_RECLOCK_LATENCY:
463                 CHECK_CONTROL_DATA_SIZE(sizeof(double));
464                 CTDB_UPDATE_RECLOCK_LATENCY(ctdb, "recd reclock", reclock.recd, *((double *)indata.dptr));
465                 return 0;
466         case CTDB_CONTROL_GET_RECLOCK_FILE:
467                 CHECK_CONTROL_DATA_SIZE(0);
468                 if (ctdb->recovery_lock_file != NULL) {
469                         outdata->dptr  = discard_const(ctdb->recovery_lock_file);
470                         outdata->dsize = strlen(ctdb->recovery_lock_file) + 1;
471                 }
472                 return 0;
473         case CTDB_CONTROL_SET_RECLOCK_FILE:
474                 ctdb->tunable.verify_recovery_lock = 0;
475                 if (ctdb->recovery_lock_file != NULL) {
476                         talloc_free(ctdb->recovery_lock_file);
477                         ctdb->recovery_lock_file = NULL;
478                 }
479                 if (indata.dsize > 0) {
480                         ctdb->recovery_lock_file = talloc_strdup(ctdb, discard_const(indata.dptr));
481                         ctdb->tunable.verify_recovery_lock = 1;
482                 }
483                 return 0;
484
485         case CTDB_CONTROL_STOP_NODE:
486                 CHECK_CONTROL_DATA_SIZE(0);
487                 return ctdb_control_stop_node(ctdb, c, async_reply);
488
489         case CTDB_CONTROL_CONTINUE_NODE:
490                 CHECK_CONTROL_DATA_SIZE(0);
491                 return ctdb_control_continue_node(ctdb);
492
493         case CTDB_CONTROL_SET_NATGWSTATE: {
494                 uint32_t natgwstate;
495
496                 CHECK_CONTROL_DATA_SIZE(sizeof(uint32_t));              
497                 natgwstate = *(uint32_t *)indata.dptr;
498                 if (natgwstate == 0) {
499                         ctdb->capabilities &= ~CTDB_CAP_NATGW;
500                 } else {
501                         ctdb->capabilities |= CTDB_CAP_NATGW;
502                 }
503                 return 0;
504         }
505
506         case CTDB_CONTROL_SET_LMASTERROLE: {
507                 uint32_t lmasterrole;
508
509                 CHECK_CONTROL_DATA_SIZE(sizeof(uint32_t));              
510                 lmasterrole = *(uint32_t *)indata.dptr;
511                 if (lmasterrole == 0) {
512                         ctdb->capabilities &= ~CTDB_CAP_LMASTER;
513                 } else {
514                         ctdb->capabilities |= CTDB_CAP_LMASTER;
515                 }
516                 return 0;
517         }
518
519         case CTDB_CONTROL_SET_RECMASTERROLE: {
520                 uint32_t recmasterrole;
521
522                 CHECK_CONTROL_DATA_SIZE(sizeof(uint32_t));              
523                 recmasterrole = *(uint32_t *)indata.dptr;
524                 if (recmasterrole == 0) {
525                         ctdb->capabilities &= ~CTDB_CAP_RECMASTER;
526                 } else {
527                         ctdb->capabilities |= CTDB_CAP_RECMASTER;
528                 }
529                 return 0;
530         }
531
532         case CTDB_CONTROL_ENABLE_SCRIPT:
533                 return ctdb_control_enable_script(ctdb, indata);
534
535         case CTDB_CONTROL_DISABLE_SCRIPT:
536                 return ctdb_control_disable_script(ctdb, indata);
537
538         case CTDB_CONTROL_SET_BAN_STATE:
539                 CHECK_CONTROL_DATA_SIZE(sizeof(struct ctdb_ban_time));
540                 return ctdb_control_set_ban_state(ctdb, indata);
541
542         case CTDB_CONTROL_GET_BAN_STATE:
543                 CHECK_CONTROL_DATA_SIZE(0);
544                 return ctdb_control_get_ban_state(ctdb, outdata);
545
546         case CTDB_CONTROL_SET_DB_PRIORITY:
547                 CHECK_CONTROL_DATA_SIZE(sizeof(struct ctdb_db_priority));
548                 return ctdb_control_set_db_priority(ctdb, indata);
549
550         case CTDB_CONTROL_GET_DB_PRIORITY: {
551                 uint32_t db_id;
552                 struct ctdb_db_context *ctdb_db;
553
554                 CHECK_CONTROL_DATA_SIZE(sizeof(db_id));
555                 db_id = *(uint32_t *)indata.dptr;
556                 ctdb_db = find_ctdb_db(ctdb, db_id);
557                 if (ctdb_db == NULL) return -1;
558                 return ctdb_db->priority;
559         }
560
561         case CTDB_CONTROL_TRANSACTION_CANCEL:
562                 CHECK_CONTROL_DATA_SIZE(0);
563                 return ctdb_control_transaction_cancel(ctdb);
564
565         case CTDB_CONTROL_REGISTER_NOTIFY:
566                 return ctdb_control_register_notify(ctdb, client_id, indata);
567
568         case CTDB_CONTROL_DEREGISTER_NOTIFY:
569                 CHECK_CONTROL_DATA_SIZE(sizeof(struct ctdb_client_notify_deregister));
570                 return ctdb_control_deregister_notify(ctdb, client_id, indata);
571
572         case CTDB_CONTROL_GET_LOG:
573                 CHECK_CONTROL_DATA_SIZE(sizeof(struct ctdb_get_log_addr));
574                 return ctdb_control_get_log(ctdb, indata);
575
576         case CTDB_CONTROL_CLEAR_LOG:
577                 return ctdb_control_clear_log(ctdb);
578
579         case CTDB_CONTROL_GET_DB_SEQNUM:
580                 CHECK_CONTROL_DATA_SIZE(sizeof(uint64_t));
581                 return ctdb_control_get_db_seqnum(ctdb, indata, outdata);
582
583         case CTDB_CONTROL_DB_SET_HEALTHY:
584                 CHECK_CONTROL_DATA_SIZE(sizeof(uint32_t));
585                 return ctdb_control_db_set_healthy(ctdb, indata);
586
587         case CTDB_CONTROL_DB_GET_HEALTH:
588                 CHECK_CONTROL_DATA_SIZE(sizeof(uint32_t));
589                 return ctdb_control_db_get_health(ctdb, indata, outdata);
590
591         case CTDB_CONTROL_GET_PUBLIC_IP_INFO:
592                 CHECK_CONTROL_DATA_SIZE(sizeof(ctdb_sock_addr));
593                 return ctdb_control_get_public_ip_info(ctdb, c, indata, outdata);
594
595         case CTDB_CONTROL_GET_IFACES:
596                 CHECK_CONTROL_DATA_SIZE(0);
597                 return ctdb_control_get_ifaces(ctdb, c, outdata);
598
599         case CTDB_CONTROL_SET_IFACE_LINK_STATE:
600                 CHECK_CONTROL_DATA_SIZE(sizeof(struct ctdb_control_iface_info));
601                 return ctdb_control_set_iface_link(ctdb, c, indata);
602
603         case CTDB_CONTROL_GET_STAT_HISTORY:
604                 CHECK_CONTROL_DATA_SIZE(0);
605                 return ctdb_control_get_stat_history(ctdb, c, outdata);
606
607         default:
608                 DEBUG(DEBUG_CRIT,(__location__ " Unknown CTDB control opcode %u\n", opcode));
609                 return -1;
610         }
611 }
612
613 /*
614   send a reply for a ctdb control
615  */
616 void ctdb_request_control_reply(struct ctdb_context *ctdb, struct ctdb_req_control *c,
617                                 TDB_DATA *outdata, int32_t status, const char *errormsg)
618 {
619         struct ctdb_reply_control *r;
620         size_t len;
621         
622         /* some controls send no reply */
623         if (c->flags & CTDB_CTRL_FLAG_NOREPLY) {
624                 return;
625         }
626
627         len = offsetof(struct ctdb_reply_control, data) + (outdata?outdata->dsize:0);
628         if (errormsg) {
629                 len += strlen(errormsg);
630         }
631         r = ctdb_transport_allocate(ctdb, ctdb, CTDB_REPLY_CONTROL, len, struct ctdb_reply_control);
632         CTDB_NO_MEMORY_VOID(ctdb, r);
633
634         r->hdr.destnode     = c->hdr.srcnode;
635         r->hdr.reqid        = c->hdr.reqid;
636         r->status           = status;
637         r->datalen          = outdata?outdata->dsize:0;
638         if (outdata && outdata->dsize) {
639                 memcpy(&r->data[0], outdata->dptr, outdata->dsize);
640         }
641         if (errormsg) {
642                 r->errorlen = strlen(errormsg);
643                 memcpy(&r->data[r->datalen], errormsg, r->errorlen);
644         }
645
646         ctdb_queue_packet_opcode(ctdb, &r->hdr, c->opcode);     
647
648         talloc_free(r);
649 }
650
651 /*
652   called when a CTDB_REQ_CONTROL packet comes in
653 */
654 void ctdb_request_control(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
655 {
656         struct ctdb_req_control *c = (struct ctdb_req_control *)hdr;
657         TDB_DATA data, *outdata;
658         int32_t status;
659         bool async_reply = False;
660         const char *errormsg = NULL;
661
662         data.dptr = &c->data[0];
663         data.dsize = c->datalen;
664
665         outdata = talloc_zero(c, TDB_DATA);
666
667         status = ctdb_control_dispatch(ctdb, c, data, outdata, hdr->srcnode, 
668                                        &errormsg, &async_reply);
669
670         if (!async_reply) {
671                 ctdb_request_control_reply(ctdb, c, outdata, status, errormsg);
672         }
673 }
674
675 /*
676   called when a CTDB_REPLY_CONTROL packet comes in
677 */
678 void ctdb_reply_control(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
679 {
680         struct ctdb_reply_control *c = (struct ctdb_reply_control *)hdr;
681         TDB_DATA data;
682         struct ctdb_control_state *state;
683         const char *errormsg = NULL;
684
685         state = ctdb_reqid_find(ctdb, hdr->reqid, struct ctdb_control_state);
686         if (state == NULL) {
687                 DEBUG(DEBUG_ERR,("pnn %u Invalid reqid %u in ctdb_reply_control\n",
688                          ctdb->pnn, hdr->reqid));
689                 return;
690         }
691
692         if (hdr->reqid != state->reqid) {
693                 /* we found a record  but it was the wrong one */
694                 DEBUG(DEBUG_ERR, ("Dropped orphaned control reply with reqid:%u\n", hdr->reqid));
695                 return;
696         }
697
698         data.dptr = &c->data[0];
699         data.dsize = c->datalen;
700         if (c->errorlen) {
701                 errormsg = talloc_strndup(state, 
702                                           (char *)&c->data[c->datalen], c->errorlen);
703         }
704
705         /* make state a child of the packet, so it goes away when the packet
706            is freed. */
707         talloc_steal(hdr, state);
708
709         state->callback(ctdb, c->status, data, errormsg, state->private_data);
710 }
711
712 static int ctdb_control_destructor(struct ctdb_control_state *state)
713 {
714         ctdb_reqid_remove(state->ctdb, state->reqid);
715         return 0;
716 }
717
718 /*
719   handle a timeout of a control
720  */
721 static void ctdb_control_timeout(struct event_context *ev, struct timed_event *te, 
722                        struct timeval t, void *private_data)
723 {
724         struct ctdb_control_state *state = talloc_get_type(private_data, struct ctdb_control_state);
725         TALLOC_CTX *tmp_ctx = talloc_new(ev);
726
727         CTDB_INCREMENT_STAT(state->ctdb, timeouts.control);
728
729         talloc_steal(tmp_ctx, state);
730
731         state->callback(state->ctdb, -1, tdb_null,
732                         "ctdb_control timed out", 
733                         state->private_data);
734         talloc_free(tmp_ctx);
735 }
736
737
738 /*
739   send a control message to a node
740  */
741 int ctdb_daemon_send_control(struct ctdb_context *ctdb, uint32_t destnode,
742                              uint64_t srvid, uint32_t opcode, uint32_t client_id,
743                              uint32_t flags,
744                              TDB_DATA data,
745                              ctdb_control_callback_fn_t callback,
746                              void *private_data)
747 {
748         struct ctdb_req_control *c;
749         struct ctdb_control_state *state;
750         size_t len;
751
752         if (ctdb->methods == NULL) {
753                 DEBUG(DEBUG_ERR,(__location__ " Failed to send control. Transport is DOWN\n"));
754                 return -1;
755         }
756
757         if (((destnode == CTDB_BROADCAST_VNNMAP) || 
758              (destnode == CTDB_BROADCAST_ALL) ||
759              (destnode == CTDB_BROADCAST_CONNECTED)) && 
760             !(flags & CTDB_CTRL_FLAG_NOREPLY)) {
761                 DEBUG(DEBUG_CRIT,("Attempt to broadcast control without NOREPLY\n"));
762                 return -1;
763         }
764
765         if (destnode != CTDB_BROADCAST_VNNMAP && 
766             destnode != CTDB_BROADCAST_ALL && 
767             destnode != CTDB_BROADCAST_CONNECTED && 
768             (!ctdb_validate_pnn(ctdb, destnode) || 
769              (ctdb->nodes[destnode]->flags & NODE_FLAGS_DISCONNECTED))) {
770                 if (!(flags & CTDB_CTRL_FLAG_NOREPLY)) {
771                         callback(ctdb, -1, tdb_null, "ctdb_control to disconnected node", private_data);
772                 }
773                 return 0;
774         }
775
776         /* the state is made a child of private_data if possible. This means any reply
777            will be discarded if the private_data goes away */
778         state = talloc(private_data?private_data:ctdb, struct ctdb_control_state);
779         CTDB_NO_MEMORY(ctdb, state);
780
781         state->reqid = ctdb_reqid_new(ctdb, state);
782         state->callback = callback;
783         state->private_data = private_data;
784         state->ctdb = ctdb;
785         state->flags = flags;
786
787         talloc_set_destructor(state, ctdb_control_destructor);
788
789         len = offsetof(struct ctdb_req_control, data) + data.dsize;
790         c = ctdb_transport_allocate(ctdb, state, CTDB_REQ_CONTROL, len, 
791                                     struct ctdb_req_control);
792         CTDB_NO_MEMORY(ctdb, c);
793         talloc_set_name_const(c, "ctdb_req_control packet");
794
795         c->hdr.destnode     = destnode;
796         c->hdr.reqid        = state->reqid;
797         c->opcode           = opcode;
798         c->client_id        = client_id;
799         c->flags            = flags;
800         c->srvid            = srvid;
801         c->datalen          = data.dsize;
802         if (data.dsize) {
803                 memcpy(&c->data[0], data.dptr, data.dsize);
804         }
805
806         ctdb_queue_packet(ctdb, &c->hdr);       
807
808         if (flags & CTDB_CTRL_FLAG_NOREPLY) {
809                 talloc_free(state);
810                 return 0;
811         }
812
813         if (ctdb->tunable.control_timeout) {
814                 event_add_timed(ctdb->ev, state, 
815                                 timeval_current_ofs(ctdb->tunable.control_timeout, 0), 
816                                 ctdb_control_timeout, state);
817         }
818
819         talloc_free(c);
820         return 0;
821 }