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