source/cluster/ctdb/ctdb_cluster.c

   1 /*
   2    Unix SMB/CIFS implementation.
   3
   4    ctdb clustering hooks
   5
   6    Copyright (C) Andrew Tridgell 2006
   7
   8    This program is free software; you can redistribute it and/or modify
   9    it under the terms of the GNU General Public License as published by
  10    the Free Software Foundation; either version 2 of the License, or
  11    (at your option) any later version.
  12
  13    This program is distributed in the hope that it will be useful,
  14    but WITHOUT ANY WARRANTY; without even the implied warranty of
  15    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16    GNU General Public License for more details.
  17
  18    You should have received a copy of the GNU General Public License
  19    along with this program; if not, write to the Free Software
  20    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  21 */
  22
  23 #include "includes.h"
  24 #include "lib/events/events.h"
  25 #include "cluster/cluster.h"
  26 #include "system/filesys.h"
  27 #include "cluster/cluster_private.h"
  28 #include "lib/tdb/include/tdb.h"
  29 #include "include/ctdb.h"
  30 #include "db_wrap.h"
  31 #include "lib/util/dlinklist.h"
  32
  33 /* a linked list of messaging handlers, allowing incoming messages
  34    to be directed to the right messaging context */
  35 struct cluster_messaging_list {
  36         struct cluster_messaging_list *next, *prev;
  37         struct cluster_state *state;
  38         struct messaging_context *msg;
  39         struct server_id server;
  40         cluster_message_fn_t handler;
  41 };
  42
  43 struct cluster_state {
  44         struct ctdb_context *ctdb;
  45         struct cluster_messaging_list *list;
  46 };
  47
  48
  49
  50 /*
  51   return a server_id for a ctdb node
  52 */
  53 static struct server_id ctdb_id(struct cluster_ops *ops, uint32_t id)
  54 {
  55         struct cluster_state *state = ops->private;
  56         struct ctdb_context *ctdb = state->ctdb;
  57         struct server_id server_id;
  58         server_id.node = ctdb_get_vnn(ctdb);
  59         server_id.id = id;
  60         return server_id;
  61 }
  62
  63
  64 /*
  65   return a server_id as a string
  66 */
  67 static const char *ctdb_id_string(struct cluster_ops *ops,
  68                                   TALLOC_CTX *mem_ctx, struct server_id id)
  69 {
  70         return talloc_asprintf(mem_ctx, "%u.%u", id.node, id.id);
  71 }
  72
  73 /*
  74   this is an interim method for subsystems that have not yet been
  75   converted to use the ctdb api. It opens a shared database in the
  76   cluster temporary area, using TDB_CLEAR_IF_FIRST which relies on
  77   correct operation of fcntl locks on the shared fileystem.
  78 */
  79 static struct tdb_wrap *ctdb_tdb_tmp_open(struct cluster_ops *ops,
  80                                           TALLOC_CTX *mem_ctx, const char *dbname,
  81                                           int flags)
  82 {
  83         const char *dir = lp_parm_string(-1, "ctdb", "shared data");
  84         char *path;
  85         struct tdb_wrap *w;
  86         if (dir == NULL) {
  87                 DEBUG(0,("ERROR: You must set 'ctdb:shared data' to a cluster shared path\n"));
  88                 return NULL;
  89         }
  90         path = talloc_asprintf(mem_ctx, "%s/%s", dir, dbname);
  91         w = tdb_wrap_open(mem_ctx, path, 0,
  92                           flags | TDB_CLEAR_IF_FIRST,
  93                           O_RDWR|O_CREAT, 0600);
  94         talloc_free(path);
  95         return w;
  96 }
  97
  98 /*
  99   get at the ctdb handle
 100 */
 101 static void *ctdb_backend_handle(struct cluster_ops *ops)
 102 {
 103         struct cluster_state *state = ops->private;
 104         return (void *)state->ctdb;
 105 }
 106
 107 /*
 108   dispatch incoming ctdb messages
 109 */
 110 static void ctdb_message_handler(struct ctdb_context *ctdb, uint32_t srvid,
 111                                  TDB_DATA data, void *private)
 112 {
 113         struct cluster_state *state = talloc_get_type(private, struct cluster_state);
 114         struct cluster_messaging_list *m;
 115         for (m=state->list;m;m=m->next) {
 116                 if (srvid == m->server.id) {
 117                         DATA_BLOB bdata;
 118                         bdata.data   = data.dptr;
 119                         bdata.length = data.dsize;
 120                         m->handler(m->msg, bdata);
 121                 }
 122         }
 123 }
 124
 125 /*
 126   destroy a element of messaging list (when messaging context goes away)
 127 */
 128 static int cluster_messaging_destructor(struct cluster_messaging_list *m)
 129 {
 130         DLIST_REMOVE(m->state->list, m);
 131         return 0;
 132 }
 133
 134 /*
 135   setup a handler for ctdb messages
 136 */
 137 static NTSTATUS ctdb_message_init(struct cluster_ops *ops,
 138                                   struct messaging_context *msg,
 139                                   struct server_id server,
 140                                   cluster_message_fn_t handler)
 141 {
 142         struct cluster_state *state = ops->private;
 143         struct cluster_messaging_list *m;
 144         int ret;
 145
 146         /* setup messaging handler */
 147         ret = ctdb_set_message_handler(state->ctdb, ctdb_message_handler,
 148                                        server.id, state);
 149         if (ret == -1) {
 150                 DEBUG(0,("ctdb_set_message_handler failed - %s\n",
 151                          ctdb_errstr(state->ctdb)));
 152                 exit(1);
 153         }
 154
 155         m = talloc(msg, struct cluster_messaging_list);
 156         NT_STATUS_HAVE_NO_MEMORY(m);
 157
 158         m->state   = state;
 159         m->msg     = msg;
 160         m->server  = server;
 161         m->handler = handler;
 162         DLIST_ADD(state->list, m);
 163
 164         talloc_set_destructor(m, cluster_messaging_destructor);
 165
 166         return NT_STATUS_OK;
 167 }
 168
 169 /*
 170   send a ctdb message to another node
 171 */
 172 static NTSTATUS ctdb_message_send(struct cluster_ops *ops,
 173                                   struct server_id server, DATA_BLOB *data)
 174 {
 175         struct cluster_state *state = ops->private;
 176         struct ctdb_context *ctdb = state->ctdb;
 177         TDB_DATA tdata;
 178         int ret;
 179
 180         tdata.dptr = data->data;
 181         tdata.dsize = data->length;
 182
 183         ret = ctdb_send_message(ctdb, server.node, server.id, tdata);
 184         if (ret != 0) {
 185                 return NT_STATUS_INTERNAL_DB_CORRUPTION;
 186         }
 187         return NT_STATUS_OK;
 188 }
 189
 190 static struct cluster_ops cluster_ctdb_ops = {
 191         .cluster_id           = ctdb_id,
 192         .cluster_id_string    = ctdb_id_string,
 193         .cluster_tdb_tmp_open = ctdb_tdb_tmp_open,
 194         .backend_handle       = ctdb_backend_handle,
 195         .message_init         = ctdb_message_init,
 196         .message_send         = ctdb_message_send,
 197         .private           = NULL
 198 };
 199
 200 /* initialise ctdb */
 201 void cluster_ctdb_init(struct event_context *ev)
 202 {
 203         const char *nlist;
 204         const char *address;
 205         const char *transport;
 206         struct cluster_state *state;
 207         int ret, lacount, i;
 208         const char *db_list[] = { "brlock", "opendb" };
 209
 210         nlist = lp_parm_string(-1, "ctdb", "nlist");
 211         if (nlist == NULL) return;
 212
 213         address = lp_parm_string(-1, "ctdb", "address");
 214         if (address == NULL) return;
 215
 216         transport = lp_parm_string(-1, "ctdb", "transport");
 217         if (transport == NULL) {
 218                 transport = "tcp";
 219         }
 220
 221         state = talloc(ev, struct cluster_state);
 222         if (state == NULL) goto failed;
 223
 224         state->ctdb = ctdb_init(ev);
 225         if (state->ctdb == NULL) goto failed;
 226
 227         state->list = NULL;
 228
 229         cluster_ctdb_ops.private = state;
 230
 231         ret = ctdb_set_transport(state->ctdb, transport);
 232         if (ret == -1) {
 233                 DEBUG(0,("ctdb_set_transport failed - %s\n",
 234                          ctdb_errstr(state->ctdb)));
 235                 goto failed;
 236         }
 237
 238         if (lp_parm_bool(-1, "ctdb", "selfconnect", False)) {
 239                 DEBUG(0,("Enabling ctdb selfconnect\n"));
 240                 ctdb_set_flags(state->ctdb, CTDB_FLAG_SELF_CONNECT);
 241         }
 242
 243         lacount = lp_parm_int(-1, "ctdb", "maxlacount", -1);
 244         if (lacount != -1) {
 245                 ctdb_set_max_lacount(state->ctdb, lacount);
 246         }
 247
 248         /* tell ctdb what address to listen on */
 249         ret = ctdb_set_address(state->ctdb, address);
 250         if (ret == -1) {
 251                 DEBUG(0,("ctdb_set_address failed - %s\n", ctdb_errstr(state->ctdb)));
 252                 goto failed;
 253         }
 254
 255         /* tell ctdb what nodes are available */
 256         ret = ctdb_set_nlist(state->ctdb, nlist);
 257         if (ret == -1) {
 258                 DEBUG(0,("ctdb_set_nlist failed - %s\n", ctdb_errstr(state->ctdb)));
 259                 goto failed;
 260         }
 261
 262         /* attach all the databases we will need */
 263         for (i=0;i<ARRAY_SIZE(db_list);i++) {
 264                 struct ctdb_db_context *ctdb_db;
 265                 ctdb_db = ctdb_attach(state->ctdb, db_list[i], TDB_INTERNAL,
 266                                       O_RDWR|O_CREAT|O_TRUNC, 0666);
 267                 if (ctdb_db == NULL) goto failed;
 268         }
 269
 270         /* start the protocol running */
 271         ret = ctdb_start(state->ctdb);
 272         if (ret == -1) {
 273                 DEBUG(0,("ctdb_start failed - %s\n", ctdb_errstr(state->ctdb)));
 274                 goto failed;
 275         }
 276
 277         /* wait until all nodes are connected (should not be needed
 278            outide of test code) */
 279         ctdb_connect_wait(state->ctdb);
 280
 281         cluster_set_ops(&cluster_ctdb_ops);
 282
 283         return;
 284
 285 failed:
 286         DEBUG(0,("cluster_ctdb_init failed\n"));
 287         talloc_free(state);
 288 }