data/src/status.c

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