data/src/status.c

   1 /*
   2   Copyright (C) 2010 Proxmox Server Solutions GmbH
   3
   4   This program is free software: you can redistribute it and/or modify
   5   it under the terms of the GNU Affero General Public License as published by
   6   the Free Software Foundation, either version 3 of the License, or
   7   (at your option) any later version.
   8
   9   This program is distributed in the hope that it will be useful,
  10   but WITHOUT ANY WARRANTY; without even the implied warranty of
  11   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12   GNU Affero General Public License for more details.
  13
  14   You should have received a copy of the GNU Affero General Public License
  15   along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16
  17   Author: Dietmar Maurer <dietmar@proxmox.com>
  18
  19 */
  20
  21 #define G_LOG_DOMAIN "status"
  22
  23 #ifdef HAVE_CONFIG_H
  24 #include <config.h>
  25 #endif /* HAVE_CONFIG_H */
  26
  27 #include <stdio.h>
  28 #include <stdint.h>
  29 #include <string.h>
  30 #include <errno.h>
  31 #include <glib.h>
  32 #include <sys/syslog.h>
  33 #include <rrd.h>
  34 #include <rrd_client.h>
  35 #include <time.h>
  36 #include <ctype.h>
  37
  38 #include "cfs-utils.h"
  39 #include "status.h"
  40 #include "memdb.h"
  41 #include "logger.h"
  42
  43 #define KVSTORE_CPG_GROUP_NAME "pve_kvstore_v1"
  44
  45 typedef enum {
  46         KVSTORE_MESSAGE_UPDATE = 1,
  47         KVSTORE_MESSAGE_UPDATE_COMPLETE = 2,
  48         KVSTORE_MESSAGE_LOG = 3,
  49 } kvstore_message_t;
  50
  51 static uint32_t vminfo_version_counter;
  52
  53 typedef struct {
  54         uint32_t vmid;
  55         char *nodename;
  56         int vmtype;
  57         uint32_t version;
  58 } vminfo_t;
  59
  60 typedef struct {
  61         char *key;
  62         gpointer data;
  63         size_t len;
  64         uint32_t version;
  65 } kventry_t;
  66
  67 typedef struct {
  68         char *key;
  69         gpointer data;
  70         size_t len;
  71         uint32_t time;
  72 } rrdentry_t;
  73
  74 typedef struct {
  75         char *path;
  76         uint32_t version;
  77 } memdb_change_t;
  78
  79 static memdb_change_t memdb_change_array[] = {
  80         { .path = "corosync.conf" },
  81         { .path = "corosync.conf.new" },
  82         { .path = "storage.cfg" },
  83         { .path = "user.cfg" },
  84         { .path = "domains.cfg" },
  85         { .path = "priv/shadow.cfg" },
  86         { .path = "priv/acme/plugins.cfg" },
  87         { .path = "priv/tfa.cfg" },
  88         { .path = "priv/token.cfg" },
  89         { .path = "datacenter.cfg" },
  90         { .path = "vzdump.cron" },
  91         { .path = "ha/crm_commands" },
  92         { .path = "ha/manager_status" },
  93         { .path = "ha/resources.cfg" },
  94         { .path = "ha/groups.cfg" },
  95         { .path = "ha/fence.cfg" },
  96         { .path = "status.cfg" },
  97         { .path = "replication.cfg" },
  98         { .path = "ceph.conf" },
  99         { .path = "sdn/vnets.cfg" },
 100         { .path = "sdn/zones.cfg" },
 101         { .path = "sdn/controllers.cfg" },
 102         { .path = "sdn/.version" },
 103         { .path = "virtual-guest/cpu-models.conf" },
 104 };
 105
 106 static GMutex mutex;
 107
 108 typedef struct {
 109         time_t start_time;
 110
 111         uint32_t quorate;
 112
 113         cfs_clinfo_t *clinfo;
 114         uint32_t clinfo_version;
 115
 116         GHashTable *vmlist;
 117         uint32_t vmlist_version;
 118
 119         dfsm_t *kvstore;
 120         GHashTable *kvhash;
 121         GHashTable *rrdhash;
 122         GHashTable *iphash;
 123
 124         GHashTable *memdb_changes;
 125
 126         clusterlog_t *clusterlog;
 127 } cfs_status_t;
 128
 129 static cfs_status_t cfs_status;
 130
 131 struct cfs_clnode {
 132         char *name;
 133         uint32_t nodeid;
 134         uint32_t votes;
 135         gboolean online;
 136         GHashTable *kvhash;
 137 };
 138
 139 struct cfs_clinfo {
 140         char *cluster_name;
 141         uint32_t cman_version;
 142
 143         GHashTable *nodes_byid;
 144         GHashTable *nodes_byname;
 145 };
 146
 147 static guint
 148 g_int32_hash (gconstpointer v)
 149 {
 150         return *(const uint32_t *) v;
 151 }
 152
 153 static gboolean
 154 g_int32_equal (gconstpointer v1,
 155                gconstpointer v2)
 156 {
 157         return *((const uint32_t*) v1) == *((const uint32_t*) v2);
 158 }
 159
 160 static void vminfo_free(vminfo_t *vminfo)
 161 {
 162         g_return_if_fail(vminfo != NULL);
 163
 164         if (vminfo->nodename)
 165                 g_free(vminfo->nodename);
 166
 167
 168         g_free(vminfo);
 169 }
 170
 171 static const char *vminfo_type_to_string(vminfo_t *vminfo)
 172 {
 173         if (vminfo->vmtype == VMTYPE_QEMU) {
 174                 return "qemu";
 175         } else if (vminfo->vmtype == VMTYPE_OPENVZ) {
 176                 return "openvz";
 177         } else if (vminfo->vmtype == VMTYPE_LXC) {
 178                 return "lxc";
 179         } else {
 180                 return "unknown";
 181         }
 182 }
 183
 184 static const char *vminfo_type_to_path_type(vminfo_t *vminfo)
 185 {
 186         if (vminfo->vmtype == VMTYPE_QEMU) {
 187                 return "qemu-server"; // special case..
 188         } else {
 189                 return vminfo_type_to_string(vminfo);
 190         }
 191 }
 192
 193 int vminfo_to_path(vminfo_t *vminfo, GString *path)
 194 {
 195         g_return_val_if_fail(vminfo != NULL, -1);
 196         g_return_val_if_fail(path != NULL, -1);
 197
 198         if (!vminfo->nodename)
 199                 return 0;
 200
 201         const char *type = vminfo_type_to_path_type(vminfo);
 202         g_string_printf(path, "/nodes/%s/%s/%u.conf", vminfo->nodename, type, vminfo->vmid);
 203
 204         return 1;
 205 }
 206
 207 void cfs_clnode_destroy(
 208         cfs_clnode_t *clnode)
 209 {
 210         g_return_if_fail(clnode != NULL);
 211
 212         if (clnode->kvhash)
 213                 g_hash_table_destroy(clnode->kvhash);
 214
 215         if (clnode->name)
 216                 g_free(clnode->name);
 217
 218         g_free(clnode);
 219 }
 220
 221 cfs_clnode_t *cfs_clnode_new(
 222         const char *name,
 223         uint32_t nodeid,
 224         uint32_t votes)
 225 {
 226         g_return_val_if_fail(name != NULL, NULL);
 227
 228         cfs_clnode_t *clnode = g_new0(cfs_clnode_t, 1);
 229         if (!clnode)
 230                 return NULL;
 231
 232         clnode->name = g_strdup(name);
 233         clnode->nodeid = nodeid;
 234         clnode->votes = votes;
 235
 236         return clnode;
 237 }
 238
 239 gboolean cfs_clinfo_destroy(
 240         cfs_clinfo_t *clinfo)
 241 {
 242         g_return_val_if_fail(clinfo != NULL, FALSE);
 243
 244         if (clinfo->cluster_name)
 245                 g_free(clinfo->cluster_name);
 246
 247         if (clinfo->nodes_byname)
 248                 g_hash_table_destroy(clinfo->nodes_byname);
 249
 250         if (clinfo->nodes_byid)
 251                 g_hash_table_destroy(clinfo->nodes_byid);
 252
 253         g_free(clinfo);
 254
 255         return TRUE;
 256 }
 257
 258 cfs_clinfo_t *cfs_clinfo_new(
 259         const char *cluster_name,
 260         uint32_t cman_version)
 261 {
 262         g_return_val_if_fail(cluster_name != NULL, NULL);
 263
 264         cfs_clinfo_t *clinfo = g_new0(cfs_clinfo_t, 1);
 265         if (!clinfo)
 266                 return NULL;
 267
 268         clinfo->cluster_name = g_strdup(cluster_name);
 269         clinfo->cman_version = cman_version;
 270
 271         if (!(clinfo->nodes_byid = g_hash_table_new_full(
 272                       g_int32_hash, g_int32_equal, NULL,
 273                       (GDestroyNotify)cfs_clnode_destroy)))
 274                 goto fail;
 275
 276         if (!(clinfo->nodes_byname = g_hash_table_new(g_str_hash, g_str_equal)))
 277                 goto fail;
 278
 279         return clinfo;
 280
 281 fail:
 282         cfs_clinfo_destroy(clinfo);
 283
 284         return NULL;
 285 }
 286
 287 gboolean cfs_clinfo_add_node(
 288         cfs_clinfo_t *clinfo,
 289         cfs_clnode_t *clnode)
 290 {
 291         g_return_val_if_fail(clinfo != NULL, FALSE);
 292         g_return_val_if_fail(clnode != NULL, FALSE);
 293
 294         g_hash_table_replace(clinfo->nodes_byid, &clnode->nodeid, clnode);
 295         g_hash_table_replace(clinfo->nodes_byname, clnode->name, clnode);
 296
 297         return TRUE;
 298 }
 299
 300 int
 301 cfs_create_memberlist_msg(
 302         GString *str)
 303 {
 304         g_return_val_if_fail(str != NULL, -EINVAL);
 305
 306         g_mutex_lock (&mutex);
 307
 308         g_string_append_printf(str,"{\n");
 309
 310         guint nodecount = 0;
 311
 312         cfs_clinfo_t *clinfo = cfs_status.clinfo;
 313
 314         if (clinfo && clinfo->nodes_byid)
 315                 nodecount = g_hash_table_size(clinfo->nodes_byid);
 316
 317         if (nodecount) {
 318                 g_string_append_printf(str, "\"nodename\": \"%s\",\n", cfs.nodename);
 319                 g_string_append_printf(str, "\"version\": %u,\n", cfs_status.clinfo_version);
 320
 321                 g_string_append_printf(str, "\"cluster\": { ");
 322                 g_string_append_printf(str, "\"name\": \"%s\", \"version\": %d, "
 323                                        "\"nodes\": %d, \"quorate\": %d ",
 324                                        clinfo->cluster_name, clinfo->cman_version,
 325                                        nodecount, cfs_status.quorate);
 326
 327                 g_string_append_printf(str,"},\n");
 328                 g_string_append_printf(str,"\"nodelist\": {\n");
 329
 330                 GHashTable *ht = clinfo->nodes_byid;
 331                 GHashTableIter iter;
 332                 gpointer key, value;
 333
 334                 g_hash_table_iter_init (&iter, ht);
 335
 336                 int i = 0;
 337                 while (g_hash_table_iter_next (&iter, &key, &value)) {
 338                         cfs_clnode_t *node = (cfs_clnode_t *)value;
 339                         if (i) g_string_append_printf(str, ",\n");
 340                         i++;
 341
 342                         g_string_append_printf(str, "  \"%s\": { \"id\": %d, \"online\": %d",
 343                                                node->name, node->nodeid, node->online);
 344
 345
 346                         char *ip = (char *)g_hash_table_lookup(cfs_status.iphash, node->name);
 347                         if (ip) {
 348                                 g_string_append_printf(str, ", \"ip\": \"%s\"", ip);
 349                         }
 350
 351                         g_string_append_printf(str, "}");
 352
 353                 }
 354                 g_string_append_printf(str,"\n  }\n");
 355         } else {
 356                 g_string_append_printf(str, "\"nodename\": \"%s\",\n", cfs.nodename);
 357                 g_string_append_printf(str, "\"version\": %u\n", cfs_status.clinfo_version);
 358         }
 359
 360         g_string_append_printf(str,"}\n");
 361
 362         g_mutex_unlock (&mutex);
 363
 364         return 0;
 365 }
 366
 367 static void
 368 kventry_free(kventry_t *entry)
 369 {
 370         g_return_if_fail(entry != NULL);
 371
 372         g_free(entry->key);
 373         g_free(entry->data);
 374         g_free(entry);
 375 }
 376
 377 static GHashTable *
 378 kventry_hash_new(void)
 379 {
 380         return g_hash_table_new_full(g_str_hash, g_str_equal, NULL,
 381                                      (GDestroyNotify)kventry_free);
 382 }
 383
 384 static void
 385 rrdentry_free(rrdentry_t *entry)
 386 {
 387         g_return_if_fail(entry != NULL);
 388
 389         g_free(entry->key);
 390         g_free(entry->data);
 391         g_free(entry);
 392 }
 393
 394 static GHashTable *
 395 rrdentry_hash_new(void)
 396 {
 397         return g_hash_table_new_full(g_str_hash, g_str_equal, NULL,
 398                                      (GDestroyNotify)rrdentry_free);
 399 }
 400
 401 void
 402 cfs_cluster_log_dump(GString *str, const char *user, guint max_entries)
 403 {
 404         clusterlog_dump(cfs_status.clusterlog, str, user, max_entries);
 405 }
 406
 407 void
 408 cfs_cluster_log(clog_entry_t *entry)
 409 {
 410         g_return_if_fail(entry != NULL);
 411
 412         clusterlog_insert(cfs_status.clusterlog, entry);
 413
 414         if (cfs_status.kvstore) {
 415                 struct iovec iov[1];
 416                 iov[0].iov_base = (char *)entry;
 417                 iov[0].iov_len = clog_entry_size(entry);
 418
 419                 if (dfsm_is_initialized(cfs_status.kvstore))
 420                         dfsm_send_message(cfs_status.kvstore, KVSTORE_MESSAGE_LOG, iov, 1);
 421         }
 422 }
 423
 424 void cfs_status_init(void)
 425 {
 426         g_mutex_lock (&mutex);
 427
 428         cfs_status.start_time = time(NULL);
 429
 430         cfs_status.vmlist = vmlist_hash_new();
 431
 432         cfs_status.kvhash = kventry_hash_new();
 433
 434         cfs_status.rrdhash = rrdentry_hash_new();
 435
 436         cfs_status.iphash = g_hash_table_new_full(g_str_hash, g_str_equal, g_free, g_free);
 437
 438         cfs_status.memdb_changes = g_hash_table_new(g_str_hash, g_str_equal);
 439
 440         for (int i = 0; i < G_N_ELEMENTS(memdb_change_array); i++) {
 441                 g_hash_table_replace(cfs_status.memdb_changes,
 442                                      memdb_change_array[i].path,
 443                                      &memdb_change_array[i]);
 444         }
 445
 446         cfs_status.clusterlog = clusterlog_new();
 447
 448         // fixme:
 449         clusterlog_add(cfs_status.clusterlog, "root", "cluster", getpid(),
 450                        LOG_INFO, "starting cluster log");
 451
 452         g_mutex_unlock (&mutex);
 453 }
 454
 455 void cfs_status_cleanup(void)
 456 {
 457         g_mutex_lock (&mutex);
 458
 459         cfs_status.clinfo_version++;
 460
 461         if (cfs_status.clinfo) {
 462                 cfs_clinfo_destroy(cfs_status.clinfo);
 463                 cfs_status.clinfo = NULL;
 464         }
 465
 466         if (cfs_status.vmlist) {
 467                 g_hash_table_destroy(cfs_status.vmlist);
 468                 cfs_status.vmlist = NULL;
 469         }
 470
 471         if (cfs_status.kvhash) {
 472                 g_hash_table_destroy(cfs_status.kvhash);
 473                 cfs_status.kvhash = NULL;
 474         }
 475
 476         if (cfs_status.rrdhash) {
 477                 g_hash_table_destroy(cfs_status.rrdhash);
 478                 cfs_status.rrdhash = NULL;
 479         }
 480
 481         if (cfs_status.iphash) {
 482                 g_hash_table_destroy(cfs_status.iphash);
 483                 cfs_status.iphash = NULL;
 484         }
 485
 486         if (cfs_status.clusterlog)
 487                 clusterlog_destroy(cfs_status.clusterlog);
 488
 489         g_mutex_unlock (&mutex);
 490 }
 491
 492 void cfs_status_set_clinfo(
 493         cfs_clinfo_t *clinfo)
 494 {
 495         g_return_if_fail(clinfo != NULL);
 496
 497         g_mutex_lock (&mutex);
 498
 499         cfs_status.clinfo_version++;
 500
 501         cfs_clinfo_t *old = cfs_status.clinfo;
 502
 503         cfs_status.clinfo = clinfo;
 504
 505         cfs_message("update cluster info (cluster name  %s, version = %d)",
 506                     clinfo->cluster_name, clinfo->cman_version);
 507
 508
 509         if (old && old->nodes_byid && clinfo->nodes_byid) {
 510                 /* copy kvstore */
 511                 GHashTable *ht = clinfo->nodes_byid;
 512                 GHashTableIter iter;
 513                 gpointer key, value;
 514
 515                 g_hash_table_iter_init (&iter, ht);
 516
 517                 while (g_hash_table_iter_next (&iter, &key, &value)) {
 518                         cfs_clnode_t *node = (cfs_clnode_t *)value;
 519                         cfs_clnode_t *oldnode;
 520                         if ((oldnode = g_hash_table_lookup(old->nodes_byid, key))) {
 521                                 node->online = oldnode->online;
 522                                 node->kvhash = oldnode->kvhash;
 523                                 oldnode->kvhash = NULL;
 524                         }
 525                 }
 526
 527         }
 528
 529         if (old)
 530                 cfs_clinfo_destroy(old);
 531
 532
 533         g_mutex_unlock (&mutex);
 534 }
 535
 536 static void
 537 dump_kvstore_versions(
 538         GString *str,
 539         GHashTable *kvhash,
 540         const char *nodename)
 541 {
 542         g_return_if_fail(kvhash != NULL);
 543         g_return_if_fail(str != NULL);
 544         g_return_if_fail(nodename != NULL);
 545
 546         GHashTable *ht = kvhash;
 547         GHashTableIter iter;
 548         gpointer key, value;
 549
 550         g_string_append_printf(str, "\"%s\": {\n", nodename);
 551
 552         g_hash_table_iter_init (&iter, ht);
 553
 554         int i = 0;
 555         while (g_hash_table_iter_next (&iter, &key, &value)) {
 556                 kventry_t *entry = (kventry_t *)value;
 557                 if (i) g_string_append_printf(str, ",\n");
 558                 i++;
 559                 g_string_append_printf(str,"\"%s\": %u", entry->key, entry->version);
 560         }
 561
 562         g_string_append_printf(str, "}\n");
 563 }
 564
 565 int
 566 cfs_create_version_msg(GString *str)
 567 {
 568         g_return_val_if_fail(str != NULL, -EINVAL);
 569
 570         g_mutex_lock (&mutex);
 571
 572         g_string_append_printf(str,"{\n");
 573
 574         g_string_append_printf(str, "\"starttime\": %lu,\n", (unsigned long)cfs_status.start_time);
 575
 576         g_string_append_printf(str, "\"clinfo\": %u,\n", cfs_status.clinfo_version);
 577
 578         g_string_append_printf(str, "\"vmlist\": %u,\n", cfs_status.vmlist_version);
 579
 580         for (int i = 0; i < G_N_ELEMENTS(memdb_change_array); i++) {
 581                 g_string_append_printf(str, "\"%s\": %u,\n",
 582                                        memdb_change_array[i].path,
 583                                        memdb_change_array[i].version);
 584         }
 585
 586         g_string_append_printf(str, "\"kvstore\": {\n");
 587
 588         dump_kvstore_versions(str, cfs_status.kvhash, cfs.nodename);
 589
 590         cfs_clinfo_t *clinfo = cfs_status.clinfo;
 591
 592         if (clinfo && clinfo->nodes_byid) {
 593                 GHashTable *ht = clinfo->nodes_byid;
 594                 GHashTableIter iter;
 595                 gpointer key, value;
 596
 597                 g_hash_table_iter_init (&iter, ht);
 598
 599                 while (g_hash_table_iter_next (&iter, &key, &value)) {
 600                         cfs_clnode_t *node = (cfs_clnode_t *)value;
 601                         if (!node->kvhash)
 602                                 continue;
 603                         g_string_append_printf(str, ",\n");
 604                         dump_kvstore_versions(str, node->kvhash, node->name);
 605                 }
 606         }
 607
 608         g_string_append_printf(str,"}\n");
 609
 610         g_string_append_printf(str,"}\n");
 611
 612         g_mutex_unlock (&mutex);
 613
 614         return 0;
 615 }
 616
 617 GHashTable *
 618 vmlist_hash_new(void)
 619 {
 620         return g_hash_table_new_full(g_int_hash, g_int_equal, NULL,
 621                                      (GDestroyNotify)vminfo_free);
 622 }
 623
 624 gboolean
 625 vmlist_hash_insert_vm(
 626         GHashTable *vmlist,
 627         int vmtype,
 628         guint32 vmid,
 629         const char *nodename,
 630         gboolean replace)
 631 {
 632         g_return_val_if_fail(vmlist != NULL, FALSE);
 633         g_return_val_if_fail(nodename != NULL, FALSE);
 634         g_return_val_if_fail(vmid != 0, FALSE);
 635         g_return_val_if_fail(vmtype == VMTYPE_QEMU || vmtype == VMTYPE_OPENVZ ||
 636                              vmtype == VMTYPE_LXC, FALSE);
 637
 638         if (!replace && g_hash_table_lookup(vmlist, &vmid)) {
 639                 cfs_critical("detected duplicate VMID %d", vmid);
 640                 return FALSE;
 641         }
 642
 643         vminfo_t *vminfo = g_new0(vminfo_t, 1);
 644
 645         vminfo->vmid = vmid;
 646         vminfo->vmtype = vmtype;
 647         vminfo->nodename = g_strdup(nodename);
 648
 649         vminfo->version = ++vminfo_version_counter;
 650
 651         g_hash_table_replace(vmlist, &vminfo->vmid, vminfo);
 652
 653         return TRUE;
 654 }
 655
 656 void
 657 vmlist_register_vm(
 658         int vmtype,
 659         guint32 vmid,
 660         const char *nodename)
 661 {
 662         g_return_if_fail(cfs_status.vmlist != NULL);
 663         g_return_if_fail(nodename != NULL);
 664         g_return_if_fail(vmid != 0);
 665         g_return_if_fail(vmtype == VMTYPE_QEMU || vmtype == VMTYPE_OPENVZ ||
 666                          vmtype == VMTYPE_LXC);
 667
 668         cfs_debug("vmlist_register_vm: %s/%u %d", nodename, vmid, vmtype);
 669
 670         g_mutex_lock (&mutex);
 671
 672         cfs_status.vmlist_version++;
 673
 674         vmlist_hash_insert_vm(cfs_status.vmlist, vmtype, vmid, nodename, TRUE);
 675
 676         g_mutex_unlock (&mutex);
 677 }
 678
 679 gboolean
 680 vmlist_different_vm_exists(
 681         int vmtype,
 682         guint32 vmid,
 683         const char *nodename)
 684 {
 685         g_return_val_if_fail(cfs_status.vmlist != NULL, FALSE);
 686         g_return_val_if_fail(vmid != 0, FALSE);
 687
 688         gboolean res = FALSE;
 689
 690         g_mutex_lock (&mutex);
 691
 692         vminfo_t *vminfo;
 693         if ((vminfo = (vminfo_t *)g_hash_table_lookup(cfs_status.vmlist, &vmid))) {
 694                 if (!(vminfo->vmtype == vmtype && strcmp(vminfo->nodename, nodename) == 0))
 695                         res = TRUE;
 696         }
 697         g_mutex_unlock (&mutex);
 698
 699         return res;
 700 }
 701
 702 gboolean
 703 vmlist_vm_exists(
 704         guint32 vmid)
 705 {
 706         g_return_val_if_fail(cfs_status.vmlist != NULL, FALSE);
 707         g_return_val_if_fail(vmid != 0, FALSE);
 708
 709         g_mutex_lock (&mutex);
 710
 711         gpointer res = g_hash_table_lookup(cfs_status.vmlist, &vmid);
 712
 713         g_mutex_unlock (&mutex);
 714
 715         return res != NULL;
 716 }
 717
 718 void
 719 vmlist_delete_vm(
 720         guint32 vmid)
 721 {
 722         g_return_if_fail(cfs_status.vmlist != NULL);
 723         g_return_if_fail(vmid != 0);
 724
 725         g_mutex_lock (&mutex);
 726
 727         cfs_status.vmlist_version++;
 728
 729         g_hash_table_remove(cfs_status.vmlist, &vmid);
 730
 731         g_mutex_unlock (&mutex);
 732 }
 733
 734 void cfs_status_set_vmlist(
 735         GHashTable *vmlist)
 736 {
 737         g_return_if_fail(vmlist != NULL);
 738
 739         g_mutex_lock (&mutex);
 740
 741         cfs_status.vmlist_version++;
 742
 743         if (cfs_status.vmlist)
 744                 g_hash_table_destroy(cfs_status.vmlist);
 745
 746         cfs_status.vmlist = vmlist;
 747
 748         g_mutex_unlock (&mutex);
 749 }
 750
 751 int
 752 cfs_create_vmlist_msg(GString *str)
 753 {
 754         g_return_val_if_fail(cfs_status.vmlist != NULL, -EINVAL);
 755         g_return_val_if_fail(str != NULL, -EINVAL);
 756
 757         g_mutex_lock (&mutex);
 758
 759         g_string_append_printf(str,"{\n");
 760
 761         GHashTable *ht = cfs_status.vmlist;
 762
 763         guint count = g_hash_table_size(ht);
 764
 765         if (!count) {
 766                 g_string_append_printf(str,"\"version\": %u\n", cfs_status.vmlist_version);
 767         } else {
 768                 g_string_append_printf(str,"\"version\": %u,\n", cfs_status.vmlist_version);
 769
 770                 g_string_append_printf(str,"\"ids\": {\n");
 771
 772                 GHashTableIter iter;
 773                 gpointer key, value;
 774
 775                 g_hash_table_iter_init (&iter, ht);
 776
 777                 int first = 1;
 778                 while (g_hash_table_iter_next (&iter, &key, &value)) {
 779                         vminfo_t *vminfo = (vminfo_t *)value;
 780                         const char *type = vminfo_type_to_string(vminfo);
 781
 782                         if (!first)
 783                                 g_string_append_printf(str, ",\n");
 784                         first = 0;
 785
 786                         g_string_append_printf(str,"\"%u\": { \"node\": \"%s\", \"type\": \"%s\", \"version\": %u }",
 787                                                vminfo->vmid, vminfo->nodename, type, vminfo->version);
 788                 }
 789
 790                 g_string_append_printf(str,"}\n");
 791         }
 792         g_string_append_printf(str,"\n}\n");
 793
 794         g_mutex_unlock (&mutex);
 795
 796         return 0;
 797 }
 798
 799 // checks the conf for a line starting with '$prop:' and returns the value
 800 // afterwards, whitout initial whitespace(s), we only deal with the format
 801 // restricion imposed by our perl VM config parser, main reference is
 802 // PVE::QemuServer::parse_vm_config this allows to be very fast and still
 803 // relatively simple
 804 // main restrictions used for our advantage is the properties match reges:
 805 // ($line =~ m/^([a-z][a-z_]*\d*):\s*(.+?)\s*$/) from parse_vm_config
 806 // currently we only look at the current configuration in place, i.e., *no*
 807 // snapshort and *no* pending changes
 808 static char *
 809 _get_property_value(char *conf, int conf_size, const char *prop, int prop_len)
 810 {
 811         const char *const conf_end = conf + conf_size;
 812         char *line = conf;
 813         size_t remaining_size;
 814
 815         char *next_newline = memchr(conf, '\n', conf_size);
 816         if (next_newline == NULL) {
 817                 return NULL; // valid property lines end with \n, but none in the config
 818         }
 819         *next_newline = '\0';
 820
 821         while (line != NULL) {
 822                 if (!line[0]) goto next;
 823
 824                 // snapshot or pending section start, but nothing found yet -> not found
 825                 if (line[0] == '[') return NULL;
 826                 // properties start with /^[a-z]/, so continue early if not
 827                 if (line[0] < 'a' || line[0] > 'z') goto next;
 828
 829                 int line_len = strlen(line);
 830                 if (line_len <= prop_len + 1) goto next;
 831
 832                 if (line[prop_len] == ':' && memcmp(line, prop, prop_len) == 0) { // found
 833                         char *v_start = &line[prop_len + 1];
 834
 835                         // drop initial value whitespaces here already
 836                         while (*v_start && isspace(*v_start)) v_start++;
 837
 838                         if (!*v_start) return NULL;
 839
 840                         char *v_end = &line[line_len - 1];
 841                         while (v_end > v_start && isspace(*v_end)) v_end--;
 842                         v_end[1] = '\0';
 843
 844                         return v_start;
 845                 }
 846 next:
 847                 line = next_newline + 1;
 848                 remaining_size = conf_end - line;
 849                 if (remaining_size <= prop_len) {
 850                         return NULL;
 851                 }
 852                 next_newline = memchr(line, '\n', remaining_size);
 853                 if (next_newline == NULL) {
 854                         return NULL; // valid property lines end with \n, but none in the config
 855                 }
 856                 *next_newline = '\0';
 857         }
 858
 859         return NULL; // not found
 860 }
 861
 862 static void
 863 _g_str_append_kv_jsonescaped(GString *str, const char *k, const char *v)
 864 {
 865         g_string_append_printf(str, "\"%s\": \"", k);
 866
 867         for (; *v; v++) {
 868                 if (*v == '\\' || *v == '"') {
 869                         g_string_append_c(str, '\\');
 870                 }
 871                 g_string_append_c(str, *v);
 872         }
 873
 874         g_string_append_c(str, '"');
 875 }
 876
 877 int
 878 cfs_create_guest_conf_property_msg(GString *str, memdb_t *memdb, const char *prop, uint32_t vmid)
 879 {
 880         g_return_val_if_fail(cfs_status.vmlist != NULL, -EINVAL);
 881         g_return_val_if_fail(str != NULL, -EINVAL);
 882
 883         int prop_len = strlen(prop);
 884         int res = 0;
 885         GString *path = NULL;
 886
 887         // Prelock &memdb->mutex in order to enable the usage of memdb_read_nolock
 888         // to prevent Deadlocks as in #2553
 889         g_mutex_lock (&memdb->mutex);
 890         g_mutex_lock (&mutex);
 891
 892         g_string_printf(str,"{\n");
 893
 894         GHashTable *ht = cfs_status.vmlist;
 895         gpointer tmp = NULL;
 896         if (!g_hash_table_size(ht)) {
 897                 goto ret;
 898         }
 899
 900         path = g_string_sized_new(256);
 901         if (vmid >= 100) {
 902                 vminfo_t *vminfo = (vminfo_t *) g_hash_table_lookup(cfs_status.vmlist, &vmid);
 903                 if (vminfo == NULL) goto enoent;
 904
 905                 if (!vminfo_to_path(vminfo, path)) goto err;
 906
 907                 // use memdb_read_nolock because lock is handled here
 908                 int size = memdb_read_nolock(memdb, path->str, &tmp);
 909                 if (tmp == NULL) goto err;
 910                 if (size <= prop_len) goto ret;
 911
 912                 char *val = _get_property_value(tmp, size, prop, prop_len);
 913                 if (val == NULL) goto ret;
 914
 915                 g_string_append_printf(str, "\"%u\":{", vmid);
 916                 _g_str_append_kv_jsonescaped(str, prop, val);
 917                 g_string_append_c(str, '}');
 918
 919         } else {
 920                 GHashTableIter iter;
 921                 g_hash_table_iter_init (&iter, ht);
 922
 923                 gpointer key, value;
 924                 int first = 1;
 925                 while (g_hash_table_iter_next (&iter, &key, &value)) {
 926                         vminfo_t *vminfo = (vminfo_t *)value;
 927
 928                         if (!vminfo_to_path(vminfo, path)) goto err;
 929
 930                         g_free(tmp); // no-op if already null
 931                         tmp = NULL;
 932                         // use memdb_read_nolock because lock is handled here
 933                         int size = memdb_read_nolock(memdb, path->str, &tmp);
 934                         if (tmp == NULL || size <= prop_len) continue;
 935
 936                         char *val = _get_property_value(tmp, size, prop, prop_len);
 937                         if (val == NULL) continue;
 938
 939                         if (!first) g_string_append_printf(str, ",\n");
 940                         else first = 0;
 941
 942                         g_string_append_printf(str, "\"%u\":{", vminfo->vmid);
 943                         _g_str_append_kv_jsonescaped(str, prop, val);
 944                         g_string_append_c(str, '}');
 945                 }
 946         }
 947 ret:
 948         g_free(tmp);
 949         if (path != NULL) {
 950                 g_string_free(path, TRUE);
 951         }
 952         g_string_append_printf(str,"\n}\n");
 953         g_mutex_unlock (&mutex);
 954         g_mutex_unlock (&memdb->mutex);
 955         return res;
 956 err:
 957         res = -EIO;
 958         goto ret;
 959 enoent:
 960         res = -ENOENT;
 961         goto ret;
 962 }
 963
 964 void
 965 record_memdb_change(const char *path)
 966 {
 967         g_return_if_fail(cfs_status.memdb_changes != 0);
 968
 969         memdb_change_t *ce;
 970
 971         if ((ce = (memdb_change_t *)g_hash_table_lookup(cfs_status.memdb_changes, path))) {
 972                 ce->version++;
 973         }
 974 }
 975
 976 void
 977 record_memdb_reload(void)
 978 {
 979         for (int i = 0; i < G_N_ELEMENTS(memdb_change_array); i++) {
 980                 memdb_change_array[i].version++;
 981         }
 982 }
 983
 984 static gboolean
 985 kventry_hash_set(
 986         GHashTable *kvhash,
 987         const char *key,
 988         gconstpointer data,
 989         size_t len)
 990 {
 991         g_return_val_if_fail(kvhash != NULL, FALSE);
 992         g_return_val_if_fail(key != NULL, FALSE);
 993         g_return_val_if_fail(data != NULL, FALSE);
 994
 995         kventry_t *entry;
 996         if (!len) {
 997                 g_hash_table_remove(kvhash, key);
 998         } else if ((entry = (kventry_t *)g_hash_table_lookup(kvhash, key))) {
 999                 g_free(entry->data);
1000                 entry->data = g_memdup(data, len);
1001                 entry->len = len;
1002                 entry->version++;
1003         } else {
1004                 kventry_t *entry = g_new0(kventry_t, 1);
1005
1006                 entry->key = g_strdup(key);
1007                 entry->data = g_memdup(data, len);
1008                 entry->len = len;
1009
1010                 g_hash_table_replace(kvhash, entry->key, entry);
1011         }
1012
1013         return TRUE;
1014 }
1015
1016 static const char *rrd_def_node[] = {
1017         "DS:loadavg:GAUGE:120:0:U",
1018         "DS:maxcpu:GAUGE:120:0:U",
1019         "DS:cpu:GAUGE:120:0:U",
1020         "DS:iowait:GAUGE:120:0:U",
1021         "DS:memtotal:GAUGE:120:0:U",
1022         "DS:memused:GAUGE:120:0:U",
1023         "DS:swaptotal:GAUGE:120:0:U",
1024         "DS:swapused:GAUGE:120:0:U",
1025         "DS:roottotal:GAUGE:120:0:U",
1026         "DS:rootused:GAUGE:120:0:U",
1027         "DS:netin:DERIVE:120:0:U",
1028         "DS:netout:DERIVE:120:0:U",
1029
1030         "RRA:AVERAGE:0.5:1:70", // 1 min avg - one hour
1031         "RRA:AVERAGE:0.5:30:70", // 30 min avg - one day
1032         "RRA:AVERAGE:0.5:180:70", // 3 hour avg - one week
1033         "RRA:AVERAGE:0.5:720:70", // 12 hour avg - one month
1034         "RRA:AVERAGE:0.5:10080:70", // 7 day avg - ony year
1035
1036         "RRA:MAX:0.5:1:70", // 1 min max - one hour
1037         "RRA:MAX:0.5:30:70", // 30 min max - one day
1038         "RRA:MAX:0.5:180:70",  // 3 hour max - one week
1039         "RRA:MAX:0.5:720:70", // 12 hour max - one month
1040         "RRA:MAX:0.5:10080:70", // 7 day max - ony year
1041         NULL,
1042 };
1043
1044 static const char *rrd_def_vm[] = {
1045         "DS:maxcpu:GAUGE:120:0:U",
1046         "DS:cpu:GAUGE:120:0:U",
1047         "DS:maxmem:GAUGE:120:0:U",
1048         "DS:mem:GAUGE:120:0:U",
1049         "DS:maxdisk:GAUGE:120:0:U",
1050         "DS:disk:GAUGE:120:0:U",
1051         "DS:netin:DERIVE:120:0:U",
1052         "DS:netout:DERIVE:120:0:U",
1053         "DS:diskread:DERIVE:120:0:U",
1054         "DS:diskwrite:DERIVE:120:0:U",
1055
1056         "RRA:AVERAGE:0.5:1:70", // 1 min avg - one hour
1057         "RRA:AVERAGE:0.5:30:70", // 30 min avg - one day
1058         "RRA:AVERAGE:0.5:180:70", // 3 hour avg - one week
1059         "RRA:AVERAGE:0.5:720:70", // 12 hour avg - one month
1060         "RRA:AVERAGE:0.5:10080:70", // 7 day avg - ony year
1061
1062         "RRA:MAX:0.5:1:70", // 1 min max - one hour
1063         "RRA:MAX:0.5:30:70", // 30 min max - one day
1064         "RRA:MAX:0.5:180:70",  // 3 hour max - one week
1065         "RRA:MAX:0.5:720:70", // 12 hour max - one month
1066         "RRA:MAX:0.5:10080:70", // 7 day max - ony year
1067         NULL,
1068 };
1069
1070 static const char *rrd_def_storage[] = {
1071         "DS:total:GAUGE:120:0:U",
1072         "DS:used:GAUGE:120:0:U",
1073
1074         "RRA:AVERAGE:0.5:1:70", // 1 min avg - one hour
1075         "RRA:AVERAGE:0.5:30:70", // 30 min avg - one day
1076         "RRA:AVERAGE:0.5:180:70", // 3 hour avg - one week
1077         "RRA:AVERAGE:0.5:720:70", // 12 hour avg - one month
1078         "RRA:AVERAGE:0.5:10080:70", // 7 day avg - ony year
1079
1080         "RRA:MAX:0.5:1:70", // 1 min max - one hour
1081         "RRA:MAX:0.5:30:70", // 30 min max - one day
1082         "RRA:MAX:0.5:180:70",  // 3 hour max - one week
1083         "RRA:MAX:0.5:720:70", // 12 hour max - one month
1084         "RRA:MAX:0.5:10080:70", // 7 day max - ony year
1085         NULL,
1086 };
1087
1088 #define RRDDIR "/var/lib/rrdcached/db"
1089
1090 static void
1091 create_rrd_file(
1092         const char *filename,
1093         int argcount,
1094         const char *rrddef[])
1095 {
1096         /* start at day boundary */
1097         time_t ctime;
1098         time(&ctime);
1099         struct tm *ltm = localtime(&ctime);
1100         ltm->tm_sec = 0;
1101         ltm->tm_min = 0;
1102         ltm->tm_hour = 0;
1103
1104         rrd_clear_error();
1105         if (rrd_create_r(filename, 60, timelocal(ltm), argcount, rrddef)) {
1106                 cfs_message("RRD create error %s: %s", filename, rrd_get_error());
1107         }
1108 }
1109
1110 static inline const char *
1111 rrd_skip_data(
1112         const char *data,
1113         int count)
1114 {
1115         int found = 0;
1116         while (*data && found < count) {
1117                 if (*data++ == ':')
1118                         found++;
1119         }
1120         return data;
1121 }
1122
1123 static void
1124 update_rrd_data(
1125         const char *key,
1126         gconstpointer data,
1127         size_t len)
1128 {
1129         g_return_if_fail(key != NULL);
1130         g_return_if_fail(data != NULL);
1131         g_return_if_fail(len > 0);
1132         g_return_if_fail(len < 4096);
1133
1134         static const char *rrdcsock = "unix:/var/run/rrdcached.sock";
1135
1136         int use_daemon = 1;
1137         if (rrdc_connect(rrdcsock) != 0)
1138                 use_daemon = 0;
1139
1140         char *filename = NULL;
1141
1142         int skip = 0;
1143
1144         if (strncmp(key, "pve2-node/", 10) == 0) {
1145                 const char *node = key + 10;
1146
1147                 skip = 2;
1148
1149                 if (strchr(node, '/') != NULL)
1150                         goto keyerror;
1151
1152                 if (strlen(node) < 1)
1153                         goto keyerror;
1154
1155                 filename = g_strdup_printf(RRDDIR "/%s", key);
1156
1157                 if (!g_file_test(filename, G_FILE_TEST_EXISTS)) {
1158
1159                         mkdir(RRDDIR "/pve2-node", 0755);
1160                         int argcount = sizeof(rrd_def_node)/sizeof(void*) - 1;
1161                         create_rrd_file(filename, argcount, rrd_def_node);
1162                 }
1163
1164         } else if ((strncmp(key, "pve2-vm/", 8) == 0) ||
1165                    (strncmp(key, "pve2.3-vm/", 10) == 0)) {
1166                 const char *vmid;
1167
1168                 if (strncmp(key, "pve2-vm/", 8) == 0) {
1169                         vmid = key + 8;
1170                         skip = 2;
1171                 } else {
1172                         vmid = key + 10;
1173                         skip = 4;
1174                 }
1175
1176                 if (strchr(vmid, '/') != NULL)
1177                         goto keyerror;
1178
1179                 if (strlen(vmid) < 1)
1180                         goto keyerror;
1181
1182                 filename = g_strdup_printf(RRDDIR "/%s/%s", "pve2-vm", vmid);
1183
1184                 if (!g_file_test(filename, G_FILE_TEST_EXISTS)) {
1185
1186                         mkdir(RRDDIR "/pve2-vm", 0755);
1187                         int argcount = sizeof(rrd_def_vm)/sizeof(void*) - 1;
1188                         create_rrd_file(filename, argcount, rrd_def_vm);
1189                 }
1190
1191         } else if (strncmp(key, "pve2-storage/", 13) == 0) {
1192                 const char *node = key + 13;
1193
1194                 const char *storage = node;
1195                 while (*storage && *storage != '/')
1196                         storage++;
1197
1198                 if (*storage != '/' || ((storage - node) < 1))
1199                         goto keyerror;
1200
1201                 storage++;
1202
1203                 if (strchr(storage, '/') != NULL)
1204                         goto keyerror;
1205
1206                 if (strlen(storage) < 1)
1207                         goto keyerror;
1208
1209                 filename = g_strdup_printf(RRDDIR "/%s", key);
1210
1211                 if (!g_file_test(filename, G_FILE_TEST_EXISTS)) {
1212
1213                         mkdir(RRDDIR "/pve2-storage", 0755);
1214
1215                         char *dir = g_path_get_dirname(filename);
1216                         mkdir(dir, 0755);
1217                         g_free(dir);
1218
1219                         int argcount = sizeof(rrd_def_storage)/sizeof(void*) - 1;
1220                         create_rrd_file(filename, argcount, rrd_def_storage);
1221                 }
1222
1223         } else {
1224                 goto keyerror;
1225         }
1226
1227         const char *dp = skip ? rrd_skip_data(data, skip) : data;
1228
1229         const char *update_args[] = { dp, NULL };
1230
1231         if (use_daemon) {
1232                 int status;
1233                 if ((status = rrdc_update(filename, 1, update_args)) != 0) {
1234                         cfs_message("RRDC update error %s: %d", filename, status);
1235                         rrdc_disconnect();
1236                         rrd_clear_error();
1237                         if (rrd_update_r(filename, NULL, 1, update_args) != 0) {
1238                                 cfs_message("RRD update error %s: %s", filename, rrd_get_error());
1239                         }
1240                 }
1241
1242         } else {
1243                 rrd_clear_error();
1244                 if (rrd_update_r(filename, NULL, 1, update_args) != 0) {
1245                         cfs_message("RRD update error %s: %s", filename, rrd_get_error());
1246                 }
1247         }
1248
1249 ret:
1250         if (filename)
1251                 g_free(filename);
1252
1253         return;
1254
1255 keyerror:
1256         cfs_critical("RRD update error: unknown/wrong key %s", key);
1257         goto ret;
1258 }
1259
1260 static gboolean
1261 rrd_entry_is_old(
1262         gpointer key,
1263         gpointer value,
1264         gpointer user_data)
1265 {
1266         rrdentry_t *entry = (rrdentry_t *)value;
1267         uint32_t ctime = GPOINTER_TO_UINT(user_data);
1268
1269         int diff = ctime - entry->time;
1270
1271         /* remove everything older than 5 minutes */
1272         int expire = 60*5;
1273
1274         return (diff > expire) ? TRUE : FALSE;
1275 }
1276
1277 static char *rrd_dump_buf = NULL;
1278 static time_t rrd_dump_last = 0;
1279
1280 void
1281 cfs_rrd_dump(GString *str)
1282 {
1283         time_t ctime;
1284
1285         g_mutex_lock (&mutex);
1286
1287         time(&ctime);
1288         if (rrd_dump_buf && (ctime - rrd_dump_last) < 2) {
1289                 g_string_assign(str, rrd_dump_buf);
1290                 g_mutex_unlock (&mutex);
1291                 return;
1292         }
1293
1294         /* remove old data */
1295         g_hash_table_foreach_remove(cfs_status.rrdhash, rrd_entry_is_old,
1296                                     GUINT_TO_POINTER(ctime));
1297
1298         g_string_set_size(str, 0);
1299
1300         GHashTableIter iter;
1301         gpointer key, value;
1302
1303         g_hash_table_iter_init (&iter, cfs_status.rrdhash);
1304
1305         while (g_hash_table_iter_next (&iter, &key, &value)) {
1306                 rrdentry_t *entry = (rrdentry_t *)value;
1307                 g_string_append(str, key);
1308                 g_string_append(str, ":");
1309                 g_string_append(str, entry->data);
1310                 g_string_append(str, "\n");
1311         }
1312
1313         g_string_append_c(str, 0); // never return undef
1314
1315         rrd_dump_last = ctime;
1316         if (rrd_dump_buf)
1317                 g_free(rrd_dump_buf);
1318         rrd_dump_buf = g_strdup(str->str);
1319
1320         g_mutex_unlock (&mutex);
1321 }
1322
1323 static gboolean
1324 nodeip_hash_set(
1325         GHashTable *iphash,
1326         const char *nodename,
1327         const char *ip,
1328         size_t len)
1329 {
1330         g_return_val_if_fail(iphash != NULL, FALSE);
1331         g_return_val_if_fail(nodename != NULL, FALSE);
1332         g_return_val_if_fail(ip != NULL, FALSE);
1333         g_return_val_if_fail(len > 0, FALSE);
1334         g_return_val_if_fail(len < 256, FALSE);
1335         g_return_val_if_fail(ip[len-1] == 0, FALSE);
1336
1337         char *oldip = (char *)g_hash_table_lookup(iphash, nodename);
1338
1339         if (!oldip || (strcmp(oldip, ip) != 0)) {
1340                 cfs_status.clinfo_version++;
1341                 g_hash_table_replace(iphash, g_strdup(nodename), g_strdup(ip));
1342         }
1343
1344         return TRUE;
1345 }
1346
1347 static gboolean
1348 rrdentry_hash_set(
1349         GHashTable *rrdhash,
1350         const char *key,
1351         const char *data,
1352         size_t len)
1353 {
1354         g_return_val_if_fail(rrdhash != NULL, FALSE);
1355         g_return_val_if_fail(key != NULL, FALSE);
1356         g_return_val_if_fail(data != NULL, FALSE);
1357         g_return_val_if_fail(len > 0, FALSE);
1358         g_return_val_if_fail(len < 4096, FALSE);
1359         g_return_val_if_fail(data[len-1] == 0, FALSE);
1360
1361         rrdentry_t *entry;
1362         if ((entry = (rrdentry_t *)g_hash_table_lookup(rrdhash, key))) {
1363                 g_free(entry->data);
1364                 entry->data = g_memdup(data, len);
1365                 entry->len = len;
1366                 entry->time = time(NULL);
1367         } else {
1368                 rrdentry_t *entry = g_new0(rrdentry_t, 1);
1369
1370                 entry->key = g_strdup(key);
1371                 entry->data = g_memdup(data, len);
1372                 entry->len = len;
1373                 entry->time = time(NULL);
1374
1375                 g_hash_table_replace(rrdhash, entry->key, entry);
1376         }
1377
1378         update_rrd_data(key, data, len);
1379
1380         return TRUE;
1381 }
1382
1383 static int
1384 kvstore_send_update_message(
1385         dfsm_t *dfsm,
1386         const char *key,
1387         gpointer data,
1388         guint32 len)
1389 {
1390         if (!dfsm_is_initialized(dfsm))
1391                 return -EACCES;
1392
1393         struct iovec iov[2];
1394
1395         char name[256];
1396         g_strlcpy(name, key, sizeof(name));
1397
1398         iov[0].iov_base = &name;
1399         iov[0].iov_len = sizeof(name);
1400
1401         iov[1].iov_base = (char *)data;
1402         iov[1].iov_len = len;
1403
1404         if (dfsm_send_message(dfsm, KVSTORE_MESSAGE_UPDATE, iov, 2) == CS_OK)
1405                 return 0;
1406
1407         return -EACCES;
1408 }
1409
1410 static clog_entry_t *
1411 kvstore_parse_log_message(
1412         const void *msg,
1413         size_t msg_len)
1414 {
1415         g_return_val_if_fail(msg != NULL, NULL);
1416
1417         if (msg_len < sizeof(clog_entry_t)) {
1418                 cfs_critical("received short log message (%zu < %zu)", msg_len, sizeof(clog_entry_t));
1419                 return NULL;
1420         }
1421
1422         clog_entry_t *entry = (clog_entry_t *)msg;
1423
1424         uint32_t size = sizeof(clog_entry_t) + entry->node_len +
1425                 entry->ident_len + entry->tag_len + entry->msg_len;
1426
1427         if (msg_len != size) {
1428                 cfs_critical("received log message with wrong size (%zu != %u)", msg_len, size);
1429                 return NULL;
1430         }
1431
1432         char *msgptr = entry->data;
1433
1434         if (*((char *)msgptr + entry->node_len - 1)) {
1435                 cfs_critical("unterminated string in log message");
1436                 return NULL;
1437         }
1438         msgptr += entry->node_len;
1439
1440         if (*((char *)msgptr + entry->ident_len - 1)) {
1441                 cfs_critical("unterminated string in log message");
1442                 return NULL;
1443         }
1444         msgptr += entry->ident_len;
1445
1446         if (*((char *)msgptr + entry->tag_len - 1)) {
1447                 cfs_critical("unterminated string in log message");
1448                 return NULL;
1449         }
1450         msgptr += entry->tag_len;
1451
1452         if (*((char *)msgptr + entry->msg_len - 1)) {
1453                 cfs_critical("unterminated string in log message");
1454                 return NULL;
1455         }
1456
1457         return entry;
1458 }
1459
1460 static gboolean
1461 kvstore_parse_update_message(
1462         const void *msg,
1463         size_t msg_len,
1464         const char **key,
1465         gconstpointer *data,
1466         guint32 *len)
1467 {
1468         g_return_val_if_fail(msg != NULL, FALSE);
1469         g_return_val_if_fail(key != NULL, FALSE);
1470         g_return_val_if_fail(data != NULL, FALSE);
1471         g_return_val_if_fail(len != NULL, FALSE);
1472
1473         if (msg_len < 256) {
1474                 cfs_critical("received short kvstore message (%zu < 256)", msg_len);
1475                 return FALSE;
1476         }
1477
1478         /* test if key is null terminated */
1479         int i = 0;
1480         for (i = 0; i < 256; i++)
1481                 if (((char *)msg)[i] == 0)
1482                         break;
1483
1484         if (i == 256)
1485                 return FALSE;
1486
1487
1488         *len = msg_len - 256;
1489         *key = msg;
1490         *data = (char *) msg + 256;
1491
1492         return TRUE;
1493 }
1494
1495 int
1496 cfs_create_status_msg(
1497         GString *str,
1498         const char *nodename,
1499         const char *key)
1500 {
1501         g_return_val_if_fail(str != NULL, -EINVAL);
1502         g_return_val_if_fail(key != NULL, -EINVAL);
1503
1504         int res = -ENOENT;
1505
1506         GHashTable *kvhash = NULL;
1507
1508         g_mutex_lock (&mutex);
1509
1510         if (!nodename || !nodename[0] || !strcmp(nodename, cfs.nodename)) {
1511                 kvhash = cfs_status.kvhash;
1512         } else if (cfs_status.clinfo && cfs_status.clinfo->nodes_byname) {
1513                 cfs_clnode_t *clnode;
1514                 if ((clnode = g_hash_table_lookup(cfs_status.clinfo->nodes_byname, nodename)))
1515                         kvhash = clnode->kvhash;
1516         }
1517
1518         kventry_t *entry;
1519         if (kvhash && (entry = (kventry_t *)g_hash_table_lookup(kvhash, key))) {
1520                 g_string_append_len(str, entry->data, entry->len);
1521                 res = 0;
1522         }
1523
1524         g_mutex_unlock (&mutex);
1525
1526         return res;
1527 }
1528
1529 int
1530 cfs_status_set(
1531         const char *key,
1532         gpointer data,
1533         size_t len)
1534 {
1535         g_return_val_if_fail(key != NULL, FALSE);
1536         g_return_val_if_fail(data != NULL, FALSE);
1537         g_return_val_if_fail(cfs_status.kvhash != NULL, FALSE);
1538
1539         if (len > CFS_MAX_STATUS_SIZE)
1540                 return -EFBIG;
1541
1542         g_mutex_lock (&mutex);
1543
1544         gboolean res;
1545
1546         if (strncmp(key, "rrd/", 4) == 0) {
1547                 res = rrdentry_hash_set(cfs_status.rrdhash, key + 4, data, len);
1548         } else if (!strcmp(key, "nodeip")) {
1549                 res = nodeip_hash_set(cfs_status.iphash, cfs.nodename, data, len);
1550         } else {
1551                 res = kventry_hash_set(cfs_status.kvhash, key, data, len);
1552         }
1553         g_mutex_unlock (&mutex);
1554
1555         if (cfs_status.kvstore)
1556                 kvstore_send_update_message(cfs_status.kvstore, key, data, len);
1557
1558         return res ? 0 : -ENOMEM;
1559 }
1560
1561 gboolean
1562 cfs_kvstore_node_set(
1563         uint32_t nodeid,
1564         const char *key,
1565         gconstpointer data,
1566         size_t len)
1567 {
1568         g_return_val_if_fail(nodeid != 0, FALSE);
1569         g_return_val_if_fail(key != NULL, FALSE);
1570         g_return_val_if_fail(data != NULL, FALSE);
1571
1572         g_mutex_lock (&mutex);
1573
1574         if (!cfs_status.clinfo || !cfs_status.clinfo->nodes_byid)
1575                 goto ret; /* ignore */
1576
1577         cfs_clnode_t *clnode = g_hash_table_lookup(cfs_status.clinfo->nodes_byid, &nodeid);
1578         if (!clnode)
1579                 goto ret; /* ignore */
1580
1581         cfs_debug("got node %d status update %s", nodeid, key);
1582
1583         if (strncmp(key, "rrd/", 4) == 0) {
1584                 rrdentry_hash_set(cfs_status.rrdhash, key + 4, data, len);
1585         } else if (!strcmp(key, "nodeip")) {
1586                 nodeip_hash_set(cfs_status.iphash, clnode->name, data, len);
1587         } else {
1588                 if (!clnode->kvhash) {
1589                         if (!(clnode->kvhash = kventry_hash_new())) {
1590                                 goto ret; /*ignore */
1591                         }
1592                 }
1593
1594                 kventry_hash_set(clnode->kvhash, key, data, len);
1595
1596         }
1597 ret:
1598         g_mutex_unlock (&mutex);
1599
1600         return TRUE;
1601 }
1602
1603 static gboolean
1604 cfs_kvstore_sync(void)
1605 {
1606         g_return_val_if_fail(cfs_status.kvhash != NULL, FALSE);
1607         g_return_val_if_fail(cfs_status.kvstore != NULL, FALSE);
1608
1609         gboolean res = TRUE;
1610
1611         g_mutex_lock (&mutex);
1612
1613         GHashTable *ht = cfs_status.kvhash;
1614         GHashTableIter iter;
1615         gpointer key, value;
1616
1617         g_hash_table_iter_init (&iter, ht);
1618
1619         while (g_hash_table_iter_next (&iter, &key, &value)) {
1620                 kventry_t *entry = (kventry_t *)value;
1621                 kvstore_send_update_message(cfs_status.kvstore, entry->key, entry->data, entry->len);
1622         }
1623
1624         g_mutex_unlock (&mutex);
1625
1626         return res;
1627 }
1628
1629 static int
1630 dfsm_deliver(
1631         dfsm_t *dfsm,
1632         gpointer data,
1633         int *res_ptr,
1634         uint32_t nodeid,
1635         uint32_t pid,
1636         uint16_t msg_type,
1637         uint32_t msg_time,
1638         const void *msg,
1639         size_t msg_len)
1640 {
1641         g_return_val_if_fail(dfsm != NULL, -1);
1642         g_return_val_if_fail(msg != NULL, -1);
1643         g_return_val_if_fail(res_ptr != NULL, -1);
1644
1645         /* ignore message for ourself */
1646         if (dfsm_nodeid_is_local(dfsm, nodeid, pid))
1647                 goto ret;
1648
1649         if (msg_type == KVSTORE_MESSAGE_UPDATE) {
1650                 const char *key;
1651                 gconstpointer data;
1652                 guint32 len;
1653                 if (kvstore_parse_update_message(msg, msg_len, &key, &data, &len)) {
1654                         cfs_kvstore_node_set(nodeid, key, data, len);
1655                 } else {
1656                         cfs_critical("cant parse update message");
1657                 }
1658         } else if (msg_type == KVSTORE_MESSAGE_LOG) {
1659                 cfs_message("received log"); // fixme: remove
1660                 const clog_entry_t *entry;
1661                 if ((entry = kvstore_parse_log_message(msg, msg_len))) {
1662                         clusterlog_insert(cfs_status.clusterlog, entry);
1663                 } else {
1664                         cfs_critical("cant parse log message");
1665                 }
1666         } else {
1667                 cfs_critical("received unknown message type %d\n", msg_type);
1668                 goto fail;
1669         }
1670
1671 ret:
1672         *res_ptr = 0;
1673         return 1;
1674
1675 fail:
1676         *res_ptr = -EACCES;
1677         return 1;
1678 }
1679
1680 static void
1681 dfsm_confchg(
1682         dfsm_t *dfsm,
1683         gpointer data,
1684         const struct cpg_address *member_list,
1685         size_t member_list_entries)
1686 {
1687         g_return_if_fail(dfsm != NULL);
1688         g_return_if_fail(member_list != NULL);
1689
1690         cfs_debug("enter %s", __func__);
1691
1692         g_mutex_lock (&mutex);
1693
1694         cfs_clinfo_t *clinfo = cfs_status.clinfo;
1695
1696         if (clinfo && clinfo->nodes_byid) {
1697
1698                 GHashTable *ht = clinfo->nodes_byid;
1699                 GHashTableIter iter;
1700                 gpointer key, value;
1701
1702                 g_hash_table_iter_init (&iter, ht);
1703
1704                 while (g_hash_table_iter_next (&iter, &key, &value)) {
1705                         cfs_clnode_t *node = (cfs_clnode_t *)value;
1706                         node->online = FALSE;
1707                 }
1708
1709                 for (int i = 0; i < member_list_entries; i++) {
1710                         cfs_clnode_t *node;
1711                         if ((node = g_hash_table_lookup(clinfo->nodes_byid, &member_list[i].nodeid))) {
1712                                 node->online = TRUE;
1713                         }
1714                 }
1715
1716                 cfs_status.clinfo_version++;
1717         }
1718
1719         g_mutex_unlock (&mutex);
1720 }
1721
1722 static gpointer
1723 dfsm_get_state(
1724         dfsm_t *dfsm,
1725         gpointer data,
1726         unsigned int *res_len)
1727 {
1728         g_return_val_if_fail(dfsm != NULL, NULL);
1729
1730         gpointer msg = clusterlog_get_state(cfs_status.clusterlog, res_len);
1731
1732         return msg;
1733 }
1734
1735 static int
1736 dfsm_process_update(
1737         dfsm_t *dfsm,
1738         gpointer data,
1739         dfsm_sync_info_t *syncinfo,
1740         uint32_t nodeid,
1741         uint32_t pid,
1742         const void *msg,
1743         size_t msg_len)
1744 {
1745         cfs_critical("%s: received unexpected update message", __func__);
1746
1747         return -1;
1748 }
1749
1750 static int
1751 dfsm_process_state_update(
1752         dfsm_t *dfsm,
1753         gpointer data,
1754         dfsm_sync_info_t *syncinfo)
1755 {
1756         g_return_val_if_fail(dfsm != NULL, -1);
1757         g_return_val_if_fail(syncinfo != NULL, -1);
1758
1759         clog_base_t *clog[syncinfo->node_count];
1760
1761         int local_index = -1;
1762         for (int i = 0; i < syncinfo->node_count; i++) {
1763                 dfsm_node_info_t *ni = &syncinfo->nodes[i];
1764                 ni->synced = 1;
1765
1766                 if (syncinfo->local == ni)
1767                         local_index = i;
1768
1769                 clog_base_t *base = (clog_base_t *)ni->state;
1770                 if (ni->state_len > 8 && ni->state_len == clog_size(base)) {
1771                         clog[i] = ni->state;
1772                 } else {
1773                         cfs_critical("received log with wrong size %u", ni->state_len);
1774                         clog[i] = NULL;
1775                 }
1776         }
1777
1778         if (!clusterlog_merge(cfs_status.clusterlog, clog, syncinfo->node_count, local_index)) {
1779                 cfs_critical("unable to merge log files");
1780         }
1781
1782         cfs_kvstore_sync();
1783
1784         return 1;
1785 }
1786
1787 static int
1788 dfsm_commit(
1789         dfsm_t *dfsm,
1790         gpointer data,
1791         dfsm_sync_info_t *syncinfo)
1792 {
1793         g_return_val_if_fail(dfsm != NULL, -1);
1794         g_return_val_if_fail(syncinfo != NULL, -1);
1795
1796         return 1;
1797 }
1798
1799 static void
1800 dfsm_synced(dfsm_t *dfsm)
1801 {
1802         g_return_if_fail(dfsm != NULL);
1803
1804         char *ip = (char *)g_hash_table_lookup(cfs_status.iphash, cfs.nodename);
1805         if (!ip)
1806                 ip = cfs.ip;
1807
1808         cfs_status_set("nodeip", ip, strlen(ip) + 1);
1809 }
1810
1811 static int
1812 dfsm_cleanup(
1813         dfsm_t *dfsm,
1814         gpointer data,
1815         dfsm_sync_info_t *syncinfo)
1816 {
1817         return 1;
1818 }
1819
1820 static dfsm_callbacks_t kvstore_dfsm_callbacks = {
1821         .dfsm_deliver_fn = dfsm_deliver,
1822         .dfsm_confchg_fn = dfsm_confchg,
1823
1824         .dfsm_get_state_fn = dfsm_get_state,
1825         .dfsm_process_state_update_fn = dfsm_process_state_update,
1826         .dfsm_process_update_fn = dfsm_process_update,
1827         .dfsm_commit_fn = dfsm_commit,
1828         .dfsm_cleanup_fn = dfsm_cleanup,
1829         .dfsm_synced_fn = dfsm_synced,
1830 };
1831
1832 dfsm_t *
1833 cfs_status_dfsm_new(void)
1834 {
1835         g_mutex_lock (&mutex);
1836
1837         cfs_status.kvstore = dfsm_new(NULL, KVSTORE_CPG_GROUP_NAME, G_LOG_DOMAIN,
1838                                       0, &kvstore_dfsm_callbacks);
1839         g_mutex_unlock (&mutex);
1840
1841         return cfs_status.kvstore;
1842 }
1843
1844 gboolean
1845 cfs_is_quorate(void)
1846 {
1847         g_mutex_lock (&mutex);
1848         gboolean res =  cfs_status.quorate;
1849         g_mutex_unlock (&mutex);
1850
1851         return res;
1852 }
1853
1854 void
1855 cfs_set_quorate(
1856         uint32_t quorate,
1857         gboolean quiet)
1858 {
1859         g_mutex_lock (&mutex);
1860
1861         uint32_t prev_quorate = cfs_status.quorate;
1862         cfs_status.quorate = quorate;
1863
1864         if (!prev_quorate && cfs_status.quorate) {
1865                 if (!quiet)
1866                         cfs_message("node has quorum");
1867         }
1868
1869         if (prev_quorate && !cfs_status.quorate) {
1870                 if (!quiet)
1871                         cfs_message("node lost quorum");
1872         }
1873
1874         g_mutex_unlock (&mutex);
1875 }