data/src/status.c

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