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