data/src/status.c

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