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 ||
 585                              vmtype == VMTYPE_LXC, FALSE);
 586
 587         if (!replace && g_hash_table_lookup(vmlist, &vmid)) {
 588                 cfs_critical("detected duplicate VMID %d", vmid);
 589                 return FALSE;
 590         }
 591
 592         vminfo_t *vminfo = g_new0(vminfo_t, 1);
 593
 594         vminfo->vmid = vmid;
 595         vminfo->vmtype = vmtype;
 596         vminfo->nodename = g_strdup(nodename);
 597
 598         vminfo->version = ++vminfo_version_counter;
 599
 600         g_hash_table_replace(vmlist, &vminfo->vmid, vminfo);
 601
 602         return TRUE;
 603 }
 604
 605 void
 606 vmlist_register_vm(
 607         int vmtype,
 608         guint32 vmid,
 609         const char *nodename)
 610 {
 611         g_return_if_fail(cfs_status.vmlist != NULL);
 612         g_return_if_fail(nodename != NULL);
 613         g_return_if_fail(vmid != 0);
 614         g_return_if_fail(vmtype == VMTYPE_QEMU || vmtype == VMTYPE_OPENVZ ||
 615                          vmtype == VMTYPE_LXC);
 616
 617         cfs_debug("vmlist_register_vm: %s/%u %d", nodename, vmid, vmtype);
 618
 619         g_mutex_lock (&mutex);
 620
 621         cfs_status.vmlist_version++;
 622
 623         vmlist_hash_insert_vm(cfs_status.vmlist, vmtype, vmid, nodename, TRUE);
 624
 625         g_mutex_unlock (&mutex);
 626 }
 627
 628 gboolean
 629 vmlist_different_vm_exists(
 630         int vmtype,
 631         guint32 vmid,
 632         const char *nodename)
 633 {
 634         g_return_val_if_fail(cfs_status.vmlist != NULL, FALSE);
 635         g_return_val_if_fail(vmid != 0, FALSE);
 636
 637         gboolean res = FALSE;
 638
 639         g_mutex_lock (&mutex);
 640
 641         vminfo_t *vminfo;
 642         if ((vminfo = (vminfo_t *)g_hash_table_lookup(cfs_status.vmlist, &vmid))) {
 643                 if (!(vminfo->vmtype == vmtype && strcmp(vminfo->nodename, nodename) == 0))
 644                         res = TRUE;
 645         }
 646         g_mutex_unlock (&mutex);
 647
 648         return res;
 649 }
 650
 651 gboolean
 652 vmlist_vm_exists(
 653         guint32 vmid)
 654 {
 655         g_return_val_if_fail(cfs_status.vmlist != NULL, FALSE);
 656         g_return_val_if_fail(vmid != 0, FALSE);
 657
 658         g_mutex_lock (&mutex);
 659
 660         gpointer res = g_hash_table_lookup(cfs_status.vmlist, &vmid);
 661
 662         g_mutex_unlock (&mutex);
 663
 664         return res != NULL;
 665 }
 666
 667 void
 668 vmlist_delete_vm(
 669         guint32 vmid)
 670 {
 671         g_return_if_fail(cfs_status.vmlist != NULL);
 672         g_return_if_fail(vmid != 0);
 673
 674         g_mutex_lock (&mutex);
 675
 676         cfs_status.vmlist_version++;
 677
 678         g_hash_table_remove(cfs_status.vmlist, &vmid);
 679
 680         g_mutex_unlock (&mutex);
 681 }
 682
 683 void cfs_status_set_vmlist(
 684         GHashTable *vmlist)
 685 {
 686         g_return_if_fail(vmlist != NULL);
 687
 688         g_mutex_lock (&mutex);
 689
 690         cfs_status.vmlist_version++;
 691
 692         if (cfs_status.vmlist)
 693                 g_hash_table_destroy(cfs_status.vmlist);
 694
 695         cfs_status.vmlist = vmlist;
 696
 697         g_mutex_unlock (&mutex);
 698 }
 699
 700 int
 701 cfs_create_vmlist_msg(GString *str)
 702 {
 703         g_return_val_if_fail(cfs_status.vmlist != NULL, -EINVAL);
 704         g_return_val_if_fail(str != NULL, -EINVAL);
 705
 706         g_mutex_lock (&mutex);
 707
 708         g_string_append_printf(str,"{\n");
 709
 710         GHashTable *ht = cfs_status.vmlist;
 711
 712         guint count = g_hash_table_size(ht);
 713
 714         if (!count) {
 715                 g_string_append_printf(str,"\"version\": %u\n", cfs_status.vmlist_version);
 716         } else {
 717                 g_string_append_printf(str,"\"version\": %u,\n", cfs_status.vmlist_version);
 718
 719                 g_string_append_printf(str,"\"ids\": {\n");
 720
 721                 GHashTableIter iter;
 722                 gpointer key, value;
 723
 724                 g_hash_table_iter_init (&iter, ht);
 725
 726                 int first = 1;
 727                 while (g_hash_table_iter_next (&iter, &key, &value)) {
 728                         vminfo_t *vminfo = (vminfo_t *)value;
 729                         char *type;
 730                         if (vminfo->vmtype == VMTYPE_QEMU) {
 731                                 type = "qemu";
 732                         } else if (vminfo->vmtype == VMTYPE_OPENVZ) {
 733                                 type = "openvz";
 734                         } else if (vminfo->vmtype == VMTYPE_LXC) {
 735                                 type = "lxc";
 736                         } else {
 737                                 type = "unknown";
 738                         }
 739
 740                         if (!first)
 741                                 g_string_append_printf(str, ",\n");
 742                         first = 0;
 743
 744                         g_string_append_printf(str,"\"%u\": { \"node\": \"%s\", \"type\": \"%s\", \"version\": %u }",
 745                                                vminfo->vmid, vminfo->nodename, type, vminfo->version);
 746                 }
 747
 748                 g_string_append_printf(str,"}\n");
 749         }
 750         g_string_append_printf(str,"\n}\n");
 751
 752         g_mutex_unlock (&mutex);
 753
 754         return 0;
 755 }
 756
 757 void
 758 record_memdb_change(const char *path)
 759 {
 760         g_return_if_fail(cfs_status.memdb_changes != 0);
 761
 762         memdb_change_t *ce;
 763
 764         unsigned int vmid = 0;
 765         char nodename[256];
 766         char rest[4096];
 767         if (cfs_status.vmlist &&
 768             sscanf(path, "nodes/%255[^/]/lxc/%u/%4095s", nodename, &vmid, rest) == 3) {
 769                 vminfo_t *vminfo = (vminfo_t *)g_hash_table_lookup(cfs_status.vmlist, &vmid);
 770                 if (vminfo && (vminfo->vmtype == VMTYPE_LXC && strcmp(vminfo->nodename, nodename) == 0)) {
 771                         cfs_status.vmlist_version++;
 772                         vminfo->version = ++vminfo_version_counter;
 773                 }
 774         }
 775
 776         if ((ce = (memdb_change_t *)g_hash_table_lookup(cfs_status.memdb_changes, path))) {
 777                 ce->version++;
 778         }
 779 }
 780
 781 void
 782 record_memdb_reload(void)
 783 {
 784         for (int i = 0; i < G_N_ELEMENTS(memdb_change_array); i++) {
 785                 memdb_change_array[i].version++;
 786         }
 787 }
 788
 789 static gboolean
 790 kventry_hash_set(
 791         GHashTable *kvhash,
 792         const char *key,
 793         gconstpointer data,
 794         size_t len)
 795 {
 796         g_return_val_if_fail(kvhash != NULL, FALSE);
 797         g_return_val_if_fail(key != NULL, FALSE);
 798         g_return_val_if_fail(data != NULL, FALSE);
 799
 800         kventry_t *entry;
 801         if ((entry = (kventry_t *)g_hash_table_lookup(kvhash, key))) {
 802                 g_free(entry->data);
 803                 entry->data = g_memdup(data, len);
 804                 entry->len = len;
 805                 entry->version++;
 806         } else {
 807                 kventry_t *entry = g_new0(kventry_t, 1);
 808
 809                 entry->key = g_strdup(key);
 810                 entry->data = g_memdup(data, len);
 811                 entry->len = len;
 812
 813                 g_hash_table_replace(kvhash, entry->key, entry);
 814         }
 815
 816         return TRUE;
 817 }
 818
 819 static const char *rrd_def_node[] = {
 820         "DS:loadavg:GAUGE:120:0:U",
 821         "DS:maxcpu:GAUGE:120:0:U",
 822         "DS:cpu:GAUGE:120:0:U",
 823         "DS:iowait:GAUGE:120:0:U",
 824         "DS:memtotal:GAUGE:120:0:U",
 825         "DS:memused:GAUGE:120:0:U",
 826         "DS:swaptotal:GAUGE:120:0:U",
 827         "DS:swapused:GAUGE:120:0:U",
 828         "DS:roottotal:GAUGE:120:0:U",
 829         "DS:rootused:GAUGE:120:0:U",
 830         "DS:netin:DERIVE:120:0:U",
 831         "DS:netout:DERIVE:120:0:U",
 832
 833         "RRA:AVERAGE:0.5:1:70", // 1 min avg - one hour
 834         "RRA:AVERAGE:0.5:30:70", // 30 min avg - one day
 835         "RRA:AVERAGE:0.5:180:70", // 3 hour avg - one week
 836         "RRA:AVERAGE:0.5:720:70", // 12 hour avg - one month
 837         "RRA:AVERAGE:0.5:10080:70", // 7 day avg - ony year
 838
 839         "RRA:MAX:0.5:1:70", // 1 min max - one hour
 840         "RRA:MAX:0.5:30:70", // 30 min max - one day
 841         "RRA:MAX:0.5:180:70",  // 3 hour max - one week
 842         "RRA:MAX:0.5:720:70", // 12 hour max - one month
 843         "RRA:MAX:0.5:10080:70", // 7 day max - ony year
 844         NULL,
 845 };
 846
 847 static const char *rrd_def_vm[] = {
 848         "DS:maxcpu:GAUGE:120:0:U",
 849         "DS:cpu:GAUGE:120:0:U",
 850         "DS:maxmem:GAUGE:120:0:U",
 851         "DS:mem:GAUGE:120:0:U",
 852         "DS:maxdisk:GAUGE:120:0:U",
 853         "DS:disk:GAUGE:120:0:U",
 854         "DS:netin:DERIVE:120:0:U",
 855         "DS:netout:DERIVE:120:0:U",
 856         "DS:diskread:DERIVE:120:0:U",
 857         "DS:diskwrite:DERIVE:120:0:U",
 858
 859         "RRA:AVERAGE:0.5:1:70", // 1 min avg - one hour
 860         "RRA:AVERAGE:0.5:30:70", // 30 min avg - one day
 861         "RRA:AVERAGE:0.5:180:70", // 3 hour avg - one week
 862         "RRA:AVERAGE:0.5:720:70", // 12 hour avg - one month
 863         "RRA:AVERAGE:0.5:10080:70", // 7 day avg - ony year
 864
 865         "RRA:MAX:0.5:1:70", // 1 min max - one hour
 866         "RRA:MAX:0.5:30:70", // 30 min max - one day
 867         "RRA:MAX:0.5:180:70",  // 3 hour max - one week
 868         "RRA:MAX:0.5:720:70", // 12 hour max - one month
 869         "RRA:MAX:0.5:10080:70", // 7 day max - ony year
 870         NULL,
 871 };
 872
 873 static const char *rrd_def_storage[] = {
 874         "DS:total:GAUGE:120:0:U",
 875         "DS:used:GAUGE:120:0:U",
 876
 877         "RRA:AVERAGE:0.5:1:70", // 1 min avg - one hour
 878         "RRA:AVERAGE:0.5:30:70", // 30 min avg - one day
 879         "RRA:AVERAGE:0.5:180:70", // 3 hour avg - one week
 880         "RRA:AVERAGE:0.5:720:70", // 12 hour avg - one month
 881         "RRA:AVERAGE:0.5:10080:70", // 7 day avg - ony year
 882
 883         "RRA:MAX:0.5:1:70", // 1 min max - one hour
 884         "RRA:MAX:0.5:30:70", // 30 min max - one day
 885         "RRA:MAX:0.5:180:70",  // 3 hour max - one week
 886         "RRA:MAX:0.5:720:70", // 12 hour max - one month
 887         "RRA:MAX:0.5:10080:70", // 7 day max - ony year
 888         NULL,
 889 };
 890
 891 #define RRDDIR "/var/lib/rrdcached/db"
 892
 893 static void
 894 create_rrd_file(
 895         const char *filename,
 896         int argcount,
 897         const char *rrddef[])
 898 {
 899         /* start at day boundary */
 900         time_t ctime;
 901         time(&ctime);
 902         struct tm *ltm = localtime(&ctime);
 903         ltm->tm_sec = 0;
 904         ltm->tm_min = 0;
 905         ltm->tm_hour = 0;
 906
 907         rrd_clear_error();
 908         if (rrd_create_r(filename, 60, timelocal(ltm), argcount, rrddef)) {
 909                 cfs_message("RRD create error %s: %s", filename, rrd_get_error());
 910         }
 911 }
 912
 913 static inline const char *
 914 rrd_skip_data(
 915         const char *data,
 916         int count)
 917 {
 918         int found = 0;
 919         while (*data && found < count) {
 920                 if (*data++ == ':')
 921                         found++;
 922         }
 923         return data;
 924 }
 925
 926 static void
 927 update_rrd_data(
 928         const char *key,
 929         gconstpointer data,
 930         size_t len)
 931 {
 932         g_return_if_fail(key != NULL);
 933         g_return_if_fail(data != NULL);
 934         g_return_if_fail(len > 0);
 935         g_return_if_fail(len < 4096);
 936
 937         static const char *rrdcsock = "unix:/var/run/rrdcached.sock";
 938
 939         int use_daemon = 1;
 940         if (rrdc_connect(rrdcsock) != 0)
 941                 use_daemon = 0;
 942
 943         char *filename = NULL;
 944
 945         int skip = 0;
 946
 947         if (strncmp(key, "pve2-node/", 10) == 0) {
 948                 const char *node = key + 10;
 949
 950                 skip = 2;
 951
 952                 if (strchr(node, '/') != NULL)
 953                         goto keyerror;
 954
 955                 if (strlen(node) < 1)
 956                         goto keyerror;
 957
 958                 filename = g_strdup_printf(RRDDIR "/%s", key);
 959
 960                 if (!g_file_test(filename, G_FILE_TEST_EXISTS)) {
 961
 962                         mkdir(RRDDIR "/pve2-node", 0755);
 963                         int argcount = sizeof(rrd_def_node)/sizeof(void*) - 1;
 964                         create_rrd_file(filename, argcount, rrd_def_node);
 965                 }
 966
 967         } else if ((strncmp(key, "pve2-vm/", 8) == 0) ||
 968                    (strncmp(key, "pve2.3-vm/", 10) == 0)) {
 969                 const char *vmid;
 970
 971                 if (strncmp(key, "pve2-vm/", 8) == 0) {
 972                         vmid = key + 8;
 973                         skip = 2;
 974                 } else {
 975                         vmid = key + 10;
 976                         skip = 4;
 977                 }
 978
 979                 if (strchr(vmid, '/') != NULL)
 980                         goto keyerror;
 981
 982                 if (strlen(vmid) < 1)
 983                         goto keyerror;
 984
 985                 filename = g_strdup_printf(RRDDIR "/%s/%s", "pve2-vm", vmid);
 986
 987                 if (!g_file_test(filename, G_FILE_TEST_EXISTS)) {
 988
 989                         mkdir(RRDDIR "/pve2-vm", 0755);
 990                         int argcount = sizeof(rrd_def_vm)/sizeof(void*) - 1;
 991                         create_rrd_file(filename, argcount, rrd_def_vm);
 992                 }
 993
 994         } else if (strncmp(key, "pve2-storage/", 13) == 0) {
 995                 const char *node = key + 13;
 996
 997                 const char *storage = node;
 998                 while (*storage && *storage != '/')
 999                         storage++;
1000
1001                 if (*storage != '/' || ((storage - node) < 1))
1002                         goto keyerror;
1003
1004                 storage++;
1005
1006                 if (strchr(storage, '/') != NULL)
1007                         goto keyerror;
1008
1009                 if (strlen(storage) < 1)
1010                         goto keyerror;
1011
1012                 filename = g_strdup_printf(RRDDIR "/%s", key);
1013
1014                 if (!g_file_test(filename, G_FILE_TEST_EXISTS)) {
1015
1016                         mkdir(RRDDIR "/pve2-storage", 0755);
1017
1018                         char *dir = g_path_get_dirname(filename);
1019                         mkdir(dir, 0755);
1020                         g_free(dir);
1021
1022                         int argcount = sizeof(rrd_def_storage)/sizeof(void*) - 1;
1023                         create_rrd_file(filename, argcount, rrd_def_storage);
1024                 }
1025
1026         } else {
1027                 goto keyerror;
1028         }
1029
1030         const char *dp = skip ? rrd_skip_data(data, skip) : data;
1031
1032         const char *update_args[] = { dp, NULL };
1033
1034         if (use_daemon) {
1035                 int status;
1036                 if ((status = rrdc_update(filename, 1, update_args)) != 0) {
1037                         cfs_message("RRDC update error %s: %d", filename, status);
1038                         rrdc_disconnect();
1039                         rrd_clear_error();
1040                         if (rrd_update_r(filename, NULL, 1, update_args) != 0) {
1041                                 cfs_message("RRD update error %s: %s", filename, rrd_get_error());
1042                         }
1043                 }
1044
1045         } else {
1046                 rrd_clear_error();
1047                 if (rrd_update_r(filename, NULL, 1, update_args) != 0) {
1048                         cfs_message("RRD update error %s: %s", filename, rrd_get_error());
1049                 }
1050         }
1051
1052 ret:
1053         if (filename)
1054                 g_free(filename);
1055
1056         return;
1057
1058 keyerror:
1059         cfs_critical("RRD update error: unknown/wrong key %s", key);
1060         goto ret;
1061 }
1062
1063 static gboolean
1064 rrd_entry_is_old(
1065         gpointer key,
1066         gpointer value,
1067         gpointer user_data)
1068 {
1069         rrdentry_t *entry = (rrdentry_t *)value;
1070         uint32_t ctime = GPOINTER_TO_UINT(user_data);
1071
1072         int diff = ctime - entry->time;
1073
1074         /* remove everything older than 5 minutes */
1075         int expire = 60*5;
1076
1077         return (diff > expire) ? TRUE : FALSE;
1078 }
1079
1080 static char *rrd_dump_buf = NULL;
1081 static time_t rrd_dump_last = 0;
1082
1083 void
1084 cfs_rrd_dump(GString *str)
1085 {
1086         time_t ctime;
1087         time(&ctime);
1088
1089         if (rrd_dump_buf && (ctime - rrd_dump_last) < 2) {
1090                 g_string_assign(str, rrd_dump_buf);
1091                 return;
1092         }
1093
1094         /* remove old data */
1095         g_hash_table_foreach_remove(cfs_status.rrdhash, rrd_entry_is_old,
1096                                     GUINT_TO_POINTER(ctime));
1097
1098         g_string_set_size(str, 0);
1099
1100         GHashTableIter iter;
1101         gpointer key, value;
1102
1103         g_hash_table_iter_init (&iter, cfs_status.rrdhash);
1104
1105         while (g_hash_table_iter_next (&iter, &key, &value)) {
1106                 rrdentry_t *entry = (rrdentry_t *)value;
1107                 g_string_append(str, key);
1108                 g_string_append(str, ":");
1109                 g_string_append(str, entry->data);
1110                 g_string_append(str, "\n");
1111         }
1112
1113         g_string_append_c(str, 0); // never return undef
1114
1115         rrd_dump_last = ctime;
1116         if (rrd_dump_buf)
1117                 g_free(rrd_dump_buf);
1118         rrd_dump_buf = g_strdup(str->str);
1119 }
1120
1121 static gboolean
1122 nodeip_hash_set(
1123         GHashTable *iphash,
1124         const char *nodename,
1125         const char *ip,
1126         size_t len)
1127 {
1128         g_return_val_if_fail(iphash != NULL, FALSE);
1129         g_return_val_if_fail(nodename != NULL, FALSE);
1130         g_return_val_if_fail(ip != NULL, FALSE);
1131         g_return_val_if_fail(len > 0, FALSE);
1132         g_return_val_if_fail(len < 256, FALSE);
1133         g_return_val_if_fail(ip[len-1] == 0, FALSE);
1134
1135         char *oldip = (char *)g_hash_table_lookup(iphash, nodename);
1136
1137         if (!oldip || (strcmp(oldip, ip) != 0)) {
1138                 cfs_status.clinfo_version++;
1139                 g_hash_table_replace(iphash, g_strdup(nodename), g_strdup(ip));
1140         }
1141
1142         return TRUE;
1143 }
1144
1145 static gboolean
1146 rrdentry_hash_set(
1147         GHashTable *rrdhash,
1148         const char *key,
1149         const char *data,
1150         size_t len)
1151 {
1152         g_return_val_if_fail(rrdhash != NULL, FALSE);
1153         g_return_val_if_fail(key != NULL, FALSE);
1154         g_return_val_if_fail(data != NULL, FALSE);
1155         g_return_val_if_fail(len > 0, FALSE);
1156         g_return_val_if_fail(len < 4096, FALSE);
1157         g_return_val_if_fail(data[len-1] == 0, FALSE);
1158
1159         rrdentry_t *entry;
1160         if ((entry = (rrdentry_t *)g_hash_table_lookup(rrdhash, key))) {
1161                 g_free(entry->data);
1162                 entry->data = g_memdup(data, len);
1163                 entry->len = len;
1164                 entry->time = time(NULL);
1165         } else {
1166                 rrdentry_t *entry = g_new0(rrdentry_t, 1);
1167
1168                 entry->key = g_strdup(key);
1169                 entry->data = g_memdup(data, len);
1170                 entry->len = len;
1171                 entry->time = time(NULL);
1172
1173                 g_hash_table_replace(rrdhash, entry->key, entry);
1174         }
1175
1176         update_rrd_data(key, data, len);
1177
1178         return TRUE;
1179 }
1180
1181 static int
1182 kvstore_send_update_message(
1183         dfsm_t *dfsm,
1184         const char *key,
1185         gpointer data,
1186         guint32 len)
1187 {
1188
1189         struct iovec iov[2];
1190
1191         char name[256];
1192         g_strlcpy(name, key, sizeof(name));
1193
1194         iov[0].iov_base = &name;
1195         iov[0].iov_len = sizeof(name);
1196
1197         iov[1].iov_base = (char *)data;
1198         iov[1].iov_len = len;
1199
1200         if (dfsm_send_message(dfsm, KVSTORE_MESSAGE_UPDATE, iov, 2) == CS_OK)
1201                 return 0;
1202
1203         return -EACCES;
1204 }
1205
1206 static clog_entry_t *
1207 kvstore_parse_log_message(
1208         const void *msg,
1209         size_t msg_len)
1210 {
1211         g_return_val_if_fail(msg != NULL, NULL);
1212
1213         if (msg_len < sizeof(clog_entry_t)) {
1214                 cfs_critical("received short log message (%lu < %lu)", msg_len, sizeof(clog_entry_t));
1215                 return NULL;
1216         }
1217
1218         clog_entry_t *entry = (clog_entry_t *)msg;
1219
1220         uint32_t size = sizeof(clog_entry_t) + entry->node_len +
1221                 entry->ident_len + entry->tag_len + entry->msg_len;
1222
1223         if (msg_len != size) {
1224                 cfs_critical("received log message with wrong size (%lu != %u)", msg_len, size);
1225                 return NULL;
1226         }
1227
1228         msg = entry->data;
1229
1230         if (*((char *)msg + entry->node_len - 1)) {
1231                 cfs_critical("unterminated string in log message");
1232                 return NULL;
1233         }
1234         msg += entry->node_len;
1235
1236         if (*((char *)msg + entry->ident_len - 1)) {
1237                 cfs_critical("unterminated string in log message");
1238                 return NULL;
1239         }
1240         msg += entry->ident_len;
1241
1242         if (*((char *)msg + entry->tag_len - 1)) {
1243                 cfs_critical("unterminated string in log message");
1244                 return NULL;
1245         }
1246         msg += entry->tag_len;
1247
1248         if (*((char *)msg + entry->msg_len - 1)) {
1249                 cfs_critical("unterminated string in log message");
1250                 return NULL;
1251         }
1252
1253         return entry;
1254 }
1255
1256 static gboolean
1257 kvstore_parse_update_message(
1258         const void *msg,
1259         size_t msg_len,
1260         const char **key,
1261         gconstpointer *data,
1262         guint32 *len)
1263 {
1264         g_return_val_if_fail(msg != NULL, FALSE);
1265         g_return_val_if_fail(key != NULL, FALSE);
1266         g_return_val_if_fail(data != NULL, FALSE);
1267         g_return_val_if_fail(len != NULL, FALSE);
1268
1269         if (msg_len < 256) {
1270                 cfs_critical("received short kvstore message (%lu < 256)", msg_len);
1271                 return FALSE;
1272         }
1273
1274         /* test if key is null terminated */
1275         int i = 0;
1276         for (i = 0; i < 256; i++)
1277                 if (((char *)msg)[i] == 0)
1278                         break;
1279
1280         if (i == 256)
1281                 return FALSE;
1282
1283
1284         *len = msg_len - 256;
1285         *key = msg;
1286         *data = msg + 256;
1287
1288         return TRUE;
1289 }
1290
1291 int
1292 cfs_create_status_msg(
1293         GString *str,
1294         const char *nodename,
1295         const char *key)
1296 {
1297         g_return_val_if_fail(str != NULL, -EINVAL);
1298         g_return_val_if_fail(key != NULL, -EINVAL);
1299
1300         int res = -ENOENT;
1301
1302         GHashTable *kvhash = NULL;
1303
1304         g_mutex_lock (&mutex);
1305
1306         if (!nodename || !nodename[0] || !strcmp(nodename, cfs.nodename)) {
1307                 kvhash = cfs_status.kvhash;
1308         } else {
1309                 cfs_clnode_t *clnode;
1310                 if ((clnode = g_hash_table_lookup(cfs_status.clinfo->nodes_byname, nodename)))
1311                         kvhash = clnode->kvhash;
1312         }
1313
1314         kventry_t *entry;
1315         if (kvhash && (entry = (kventry_t *)g_hash_table_lookup(kvhash, key))) {
1316                 g_string_append_len(str, entry->data, entry->len);
1317                 res = 0;
1318         }
1319
1320         g_mutex_unlock (&mutex);
1321
1322         return res;
1323 }
1324
1325 int
1326 cfs_status_set(
1327         const char *key,
1328         gpointer data,
1329         size_t len)
1330 {
1331         g_return_val_if_fail(key != NULL, FALSE);
1332         g_return_val_if_fail(data != NULL, FALSE);
1333         g_return_val_if_fail(cfs_status.kvhash != NULL, FALSE);
1334
1335         if (len > CFS_MAX_STATUS_SIZE)
1336                 return -EFBIG;
1337
1338         g_mutex_lock (&mutex);
1339
1340         gboolean res;
1341
1342         if (strncmp(key, "rrd/", 4) == 0) {
1343                 res = rrdentry_hash_set(cfs_status.rrdhash, key + 4, data, len);
1344         } else if (!strcmp(key, "nodeip")) {
1345                 res = nodeip_hash_set(cfs_status.iphash, cfs.nodename, data, len);
1346         } else {
1347                 res = kventry_hash_set(cfs_status.kvhash, key, data, len);
1348         }
1349         g_mutex_unlock (&mutex);
1350
1351         if (cfs_status.kvstore)
1352                 kvstore_send_update_message(cfs_status.kvstore, key, data, len);
1353
1354         return res ? 0 : -ENOMEM;
1355 }
1356
1357 gboolean
1358 cfs_kvstore_node_set(
1359         uint32_t nodeid,
1360         const char *key,
1361         gconstpointer data,
1362         size_t len)
1363 {
1364         g_return_val_if_fail(nodeid != 0, FALSE);
1365         g_return_val_if_fail(key != NULL, FALSE);
1366         g_return_val_if_fail(data != NULL, FALSE);
1367
1368         g_mutex_lock (&mutex);
1369
1370         if (!cfs_status.clinfo || !cfs_status.clinfo->nodes_byid)
1371                 goto ret; /* ignore */
1372
1373         cfs_clnode_t *clnode = g_hash_table_lookup(cfs_status.clinfo->nodes_byid, &nodeid);
1374         if (!clnode)
1375                 goto ret; /* ignore */
1376
1377         cfs_debug("got node %d status update %s", nodeid, key);
1378
1379         if (strncmp(key, "rrd/", 4) == 0) {
1380                 rrdentry_hash_set(cfs_status.rrdhash, key + 4, data, len);
1381         } else if (!strcmp(key, "nodeip")) {
1382                 nodeip_hash_set(cfs_status.iphash, clnode->name, data, len);
1383         } else {
1384                 if (!clnode->kvhash) {
1385                         if (!(clnode->kvhash = kventry_hash_new())) {
1386                                 goto ret; /*ignore */
1387                         }
1388                 }
1389
1390                 kventry_hash_set(clnode->kvhash, key, data, len);
1391
1392         }
1393 ret:
1394         g_mutex_unlock (&mutex);
1395
1396         return TRUE;
1397 }
1398
1399 static gboolean
1400 cfs_kvstore_sync(void)
1401 {
1402         g_return_val_if_fail(cfs_status.kvhash != NULL, FALSE);
1403         g_return_val_if_fail(cfs_status.kvstore != NULL, FALSE);
1404
1405         gboolean res = TRUE;
1406
1407         g_mutex_lock (&mutex);
1408
1409         GHashTable *ht = cfs_status.kvhash;
1410         GHashTableIter iter;
1411         gpointer key, value;
1412
1413         g_hash_table_iter_init (&iter, ht);
1414
1415         while (g_hash_table_iter_next (&iter, &key, &value)) {
1416                 kventry_t *entry = (kventry_t *)value;
1417                 kvstore_send_update_message(cfs_status.kvstore, entry->key, entry->data, entry->len);
1418         }
1419
1420         g_mutex_unlock (&mutex);
1421
1422         return res;
1423 }
1424
1425 static int
1426 dfsm_deliver(
1427         dfsm_t *dfsm,
1428         gpointer data,
1429         int *res_ptr,
1430         uint32_t nodeid,
1431         uint32_t pid,
1432         uint16_t msg_type,
1433         uint32_t msg_time,
1434         const void *msg,
1435         size_t msg_len)
1436 {
1437         g_return_val_if_fail(dfsm != NULL, -1);
1438         g_return_val_if_fail(msg != NULL, -1);
1439         g_return_val_if_fail(res_ptr != NULL, -1);
1440
1441         /* ignore message for ourself */
1442         if (dfsm_nodeid_is_local(dfsm, nodeid, pid))
1443                 goto ret;
1444
1445         if (msg_type == KVSTORE_MESSAGE_UPDATE) {
1446                 const char *key;
1447                 gconstpointer data;
1448                 guint32 len;
1449                 if (kvstore_parse_update_message(msg, msg_len, &key, &data, &len)) {
1450                         cfs_kvstore_node_set(nodeid, key, data, len);
1451                 } else {
1452                         cfs_critical("cant parse update message");
1453                 }
1454         } else if (msg_type == KVSTORE_MESSAGE_LOG) {
1455                 cfs_message("received log"); // fixme: remove
1456                 const clog_entry_t *entry;
1457                 if ((entry = kvstore_parse_log_message(msg, msg_len))) {
1458                         clusterlog_insert(cfs_status.clusterlog, entry);
1459                 } else {
1460                         cfs_critical("cant parse log message");
1461                 }
1462         } else {
1463                 cfs_critical("received unknown message type %d\n", msg_type);
1464                 goto fail;
1465         }
1466
1467 ret:
1468         *res_ptr = 0;
1469         return 1;
1470
1471 fail:
1472         *res_ptr = -EACCES;
1473         return 1;
1474 }
1475
1476 static void
1477 dfsm_confchg(
1478         dfsm_t *dfsm,
1479         gpointer data,
1480         const struct cpg_address *member_list,
1481         size_t member_list_entries)
1482 {
1483         g_return_if_fail(dfsm != NULL);
1484         g_return_if_fail(member_list != NULL);
1485
1486         cfs_debug("enter %s", __func__);
1487
1488         g_mutex_lock (&mutex);
1489
1490         cfs_clinfo_t *clinfo = cfs_status.clinfo;
1491
1492         if (clinfo && clinfo->nodes_byid) {
1493
1494                 GHashTable *ht = clinfo->nodes_byid;
1495                 GHashTableIter iter;
1496                 gpointer key, value;
1497
1498                 g_hash_table_iter_init (&iter, ht);
1499
1500                 while (g_hash_table_iter_next (&iter, &key, &value)) {
1501                         cfs_clnode_t *node = (cfs_clnode_t *)value;
1502                         node->online = FALSE;
1503                 }
1504
1505                 for (int i = 0; i < member_list_entries; i++) {
1506                         cfs_clnode_t *node;
1507                         if ((node = g_hash_table_lookup(clinfo->nodes_byid, &member_list[i].nodeid))) {
1508                                 node->online = TRUE;
1509                         }
1510                 }
1511
1512                 cfs_status.clinfo_version++;
1513         }
1514
1515         g_mutex_unlock (&mutex);
1516 }
1517
1518 static gpointer
1519 dfsm_get_state(
1520         dfsm_t *dfsm,
1521         gpointer data,
1522         unsigned int *res_len)
1523 {
1524         g_return_val_if_fail(dfsm != NULL, NULL);
1525
1526         gpointer msg = clusterlog_get_state(cfs_status.clusterlog, res_len);
1527
1528         return msg;
1529 }
1530
1531 static int
1532 dfsm_process_update(
1533         dfsm_t *dfsm,
1534         gpointer data,
1535         dfsm_sync_info_t *syncinfo,
1536         uint32_t nodeid,
1537         uint32_t pid,
1538         const void *msg,
1539         size_t msg_len)
1540 {
1541         cfs_critical("%s: received unexpected update message", __func__);
1542
1543         return -1;
1544 }
1545
1546 static int
1547 dfsm_process_state_update(
1548         dfsm_t *dfsm,
1549         gpointer data,
1550         dfsm_sync_info_t *syncinfo)
1551 {
1552         g_return_val_if_fail(dfsm != NULL, -1);
1553         g_return_val_if_fail(syncinfo != NULL, -1);
1554
1555         clog_base_t *clog[syncinfo->node_count];
1556
1557         int local_index = -1;
1558         for (int i = 0; i < syncinfo->node_count; i++) {
1559                 dfsm_node_info_t *ni = &syncinfo->nodes[i];
1560                 ni->synced = 1;
1561
1562                 if (syncinfo->local == ni)
1563                         local_index = i;
1564
1565                 clog_base_t *base = (clog_base_t *)ni->state;
1566                 if (ni->state_len > 8 && ni->state_len == clog_size(base)) {
1567                         clog[i] = ni->state;
1568                 } else {
1569                         cfs_critical("received log with wrong size %u", ni->state_len);
1570                         clog[i] = NULL;
1571                 }
1572         }
1573
1574         if (!clusterlog_merge(cfs_status.clusterlog, clog, syncinfo->node_count, local_index)) {
1575                 cfs_critical("unable to merge log files");
1576         }
1577
1578         cfs_kvstore_sync();
1579
1580         return 1;
1581 }
1582
1583 static int
1584 dfsm_commit(
1585         dfsm_t *dfsm,
1586         gpointer data,
1587         dfsm_sync_info_t *syncinfo)
1588 {
1589         g_return_val_if_fail(dfsm != NULL, -1);
1590         g_return_val_if_fail(syncinfo != NULL, -1);
1591
1592         return 1;
1593 }
1594
1595 static void
1596 dfsm_synced(dfsm_t *dfsm)
1597 {
1598         g_return_if_fail(dfsm != NULL);
1599
1600         char *ip = (char *)g_hash_table_lookup(cfs_status.iphash, cfs.nodename);
1601         if (!ip)
1602                 ip = cfs.ip;
1603
1604         cfs_status_set("nodeip", ip, strlen(ip) + 1);
1605 }
1606
1607 static int
1608 dfsm_cleanup(
1609         dfsm_t *dfsm,
1610         gpointer data,
1611         dfsm_sync_info_t *syncinfo)
1612 {
1613         return 1;
1614 }
1615
1616 static dfsm_callbacks_t kvstore_dfsm_callbacks = {
1617         .dfsm_deliver_fn = dfsm_deliver,
1618         .dfsm_confchg_fn = dfsm_confchg,
1619
1620         .dfsm_get_state_fn = dfsm_get_state,
1621         .dfsm_process_state_update_fn = dfsm_process_state_update,
1622         .dfsm_process_update_fn = dfsm_process_update,
1623         .dfsm_commit_fn = dfsm_commit,
1624         .dfsm_cleanup_fn = dfsm_cleanup,
1625         .dfsm_synced_fn = dfsm_synced,
1626 };
1627
1628 dfsm_t *
1629 cfs_status_dfsm_new(void)
1630 {
1631         g_mutex_lock (&mutex);
1632
1633         cfs_status.kvstore = dfsm_new(NULL, KVSTORE_CPG_GROUP_NAME, G_LOG_DOMAIN,
1634                                       0, &kvstore_dfsm_callbacks);
1635         g_mutex_unlock (&mutex);
1636
1637         return cfs_status.kvstore;
1638 }
1639
1640 gboolean
1641 cfs_is_quorate(void)
1642 {
1643         g_mutex_lock (&mutex);
1644         gboolean res =  cfs_status.quorate;
1645         g_mutex_unlock (&mutex);
1646
1647         return res;
1648 }
1649
1650 void
1651 cfs_set_quorate(
1652         uint32_t quorate,
1653         gboolean quiet)
1654 {
1655         g_mutex_lock (&mutex);
1656
1657         uint32_t prev_quorate = cfs_status.quorate;
1658         cfs_status.quorate = quorate;
1659
1660         if (!prev_quorate && cfs_status.quorate) {
1661                 if (!quiet)
1662                         cfs_message("node has quorum");
1663         }
1664
1665         if (prev_quorate && !cfs_status.quorate) {
1666                 if (!quiet)
1667                         cfs_message("node lost quorum");
1668         }
1669
1670         g_mutex_unlock (&mutex);
1671 }
1672