data/src/logger.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 #ifdef HAVE_CONFIG_H
  22 #include <config.h>
  23 #endif /* HAVE_CONFIG_H */
  24
  25 #define _XOPEN_SOURCE /* glibc2 needs this */
  26 #include <time.h> /* for strptime */
  27
  28 #include <unistd.h>
  29 #include <stdio.h>
  30 #include <stdint.h>
  31 #include <glib.h>
  32 #include <string.h>
  33 #include <ctype.h>
  34 #include <sys/types.h>
  35 #include <sys/stat.h>
  36 #include <fcntl.h>
  37
  38
  39 #define SYSLOG_MAX_LINE_LENGTH 8192
  40
  41 #include "cfs-utils.h"
  42 #include "logger.h"
  43
  44 /*
  45  * 64 bit FNV-1a non-zero initial basis
  46  */
  47 #define FNV1A_64_INIT ((uint64_t) 0xcbf29ce484222325ULL)
  48 /*
  49  * 64 bit Fowler/Noll/Vo FNV-1a hash code
  50  * (copied from sheepdog sources)
  51  */
  52 static inline uint64_t fnv_64a_buf(const void *buf, size_t len, uint64_t hval)
  53 {
  54         unsigned char *bp = (unsigned char *) buf;
  55         unsigned char *be = bp + len;
  56         while (bp < be) {
  57                 hval ^= (uint64_t) *bp++;
  58                 hval += (hval << 1) + (hval << 4) + (hval << 5) +
  59                         (hval << 7) + (hval << 8) + (hval << 40);
  60         }
  61         return hval;
  62 }
  63
  64 static uint32_t uid_counter = 0;
  65
  66 struct clog_base {
  67         uint32_t size;
  68         uint32_t cpos;
  69         char data[];
  70 };
  71
  72 typedef struct {
  73         uint64_t node_digest;
  74         uint32_t uid;
  75         uint32_t time;
  76 } dedup_entry_t;
  77
  78 static clog_entry_t *
  79 clog_alloc_entry(
  80         clog_base_t *clog,
  81         uint32_t size)
  82 {
  83         g_return_val_if_fail(clog != NULL, NULL);
  84         g_return_val_if_fail(size > sizeof(clog_entry_t), NULL);
  85         g_return_val_if_fail(size <= CLOG_MAX_ENTRY_SIZE, NULL);
  86
  87         uint32_t realsize = ((size + 7) & 0xfffffff8);
  88
  89         uint32_t newpos;
  90
  91         if (!clog->cpos) {
  92                 newpos = sizeof(clog_base_t);
  93         } else {
  94                 clog_entry_t *cur = (clog_entry_t *)((char *)clog + clog->cpos);
  95                 newpos = cur->next;
  96                 if ((newpos + realsize) >= clog->size) {
  97                         newpos = sizeof(clog_base_t);
  98                 }
  99         }
 100
 101         clog_entry_t *entry = (clog_entry_t *)((char *)clog + newpos);
 102
 103         entry->prev = clog->cpos;
 104         clog->cpos = newpos;
 105         entry->next = newpos + realsize;
 106
 107         return entry;
 108 }
 109
 110 static void
 111 clog_dump_entry(clog_entry_t *cur, uint32_t cpos)
 112 {
 113         g_return_if_fail(cur != NULL);
 114
 115         char *node = cur->data;
 116         char *ident = node + cur->node_len;
 117         char *tag = ident + cur->ident_len;
 118         char *msg = tag + cur->tag_len;
 119
 120         time_t lt = cur->time;
 121         char tbuf[256];
 122         strftime(tbuf, sizeof(tbuf), "%F %T", localtime(&lt));
 123         printf("cpos %05d %08x %s", cpos, cur->uid, tbuf);
 124         printf(" %s{%016zX} %s[%s{%016zX}]: %s\n", node, cur->node_digest, tag, ident, cur->ident_digest, msg);
 125
 126 }
 127
 128 void
 129 clog_dump(clog_base_t *clog)
 130 {
 131         g_return_if_fail(clog != NULL);
 132
 133         uint32_t cpos = clog->cpos;
 134
 135         while (cpos && (cpos <= clog->cpos || cpos > (clog->cpos + CLOG_MAX_ENTRY_SIZE))) {
 136                 if (cpos > (clog->size - sizeof(clog_entry_t))) {
 137                         cfs_critical("log pointer out of range!");
 138                         break;
 139                 }
 140                 clog_entry_t *cur = (clog_entry_t *)((char *)clog + cpos);
 141                 clog_dump_entry(cur, cpos);
 142                 cpos = cur->prev;
 143         }
 144 }
 145
 146 void
 147 clog_dump_json(
 148         clog_base_t *clog,
 149         GString *str,
 150         const char *ident,
 151         guint max_entries)
 152 {
 153         g_return_if_fail(clog != NULL);
 154         g_return_if_fail(str != NULL);
 155
 156         guint64 ident_digest = 0;
 157
 158         if (ident && ident[0]) {
 159                 ident_digest = fnv_64a_buf(ident, strlen(ident) + 1, FNV1A_64_INIT);
 160         }
 161
 162         uint32_t cpos = clog->cpos;
 163
 164         g_string_append_printf(str, "{\n");
 165
 166         g_string_append_printf(str, "\"data\": [\n");
 167
 168         guint count = 0;
 169         while (cpos && (cpos <= clog->cpos || cpos > (clog->cpos + CLOG_MAX_ENTRY_SIZE))) {
 170                 if (cpos > (clog->size - sizeof(clog_entry_t))) {
 171                         cfs_critical("log pointer out of range!");
 172                         break;
 173                 }
 174                 clog_entry_t *cur = (clog_entry_t *)((char *)clog + cpos);
 175
 176                 cpos = cur->prev;
 177
 178                 if (count >= max_entries)
 179                         break;
 180
 181                 if (ident_digest && ident_digest != cur->ident_digest)
 182                         continue;
 183
 184                 char *node = cur->data;
 185                 char *ident = node + cur->node_len;
 186                 char *tag = ident + cur->ident_len;
 187                 char *msg = tag + cur->tag_len;
 188
 189                 if (count)
 190                         g_string_append_printf(str, ",\n");
 191
 192                 g_string_append_printf(str, "{\"uid\": %u, \"time\": %u, \"pri\": %d, \"tag\": \"%s\", "
 193                                        "\"pid\": %u, \"node\": \"%s\", \"user\": \"%s\", "
 194                                        "\"msg\": \"%s\"}", cur->uid, cur->time, cur->priority, tag,
 195                                        cur->pid, node, ident, msg);
 196
 197                 count++;
 198
 199         }
 200
 201         if (count)
 202                 g_string_append_printf(str, "\n");
 203
 204         g_string_append_printf(str, "]\n");
 205         g_string_append_printf(str, "}\n");
 206
 207 }
 208
 209 uint32_t
 210 clog_entry_size(const clog_entry_t *entry)
 211 {
 212         g_return_val_if_fail(entry != NULL, 0);
 213
 214         return sizeof(clog_entry_t) + entry->node_len +
 215                 entry->ident_len + entry->tag_len + entry->msg_len;
 216 }
 217
 218 void
 219 clog_copy(
 220         clog_base_t *clog,
 221         const clog_entry_t *entry)
 222 {
 223         g_return_if_fail(clog != NULL);
 224         g_return_if_fail(entry != NULL);
 225
 226         uint32_t size = clog_entry_size(entry);
 227
 228         clog_entry_t *new;
 229         if ((new = clog_alloc_entry(clog, size)))
 230                 memcpy((char *)new + 8, (char *)entry + 8, size - 8);
 231 };
 232
 233 uint32_t
 234 clog_pack(
 235         clog_entry_t *entry,
 236         const char *node,
 237         const char *ident,
 238         const char *tag,
 239         uint32_t pid,
 240         time_t logtime,
 241         uint8_t priority,
 242         const char *msg)
 243 {
 244         g_return_val_if_fail(entry != NULL, 0);
 245         g_return_val_if_fail(ident != NULL, 0);
 246         g_return_val_if_fail(tag != NULL, 0);
 247         g_return_val_if_fail(msg != NULL, 0);
 248         g_return_val_if_fail(priority >= 0, 0);
 249         g_return_val_if_fail(priority < 8, 0);
 250
 251         uint8_t node_len = CFS_MIN(strlen(node) + 1, 255);
 252         uint8_t ident_len = CFS_MIN(strlen(ident) + 1, 255);
 253         uint8_t tag_len = CFS_MIN(strlen(tag) + 1, 255);
 254
 255         char *msg_start = entry->data + node_len + ident_len + tag_len;
 256         *msg_start = 0;
 257
 258         int buf_len = CLOG_MAX_ENTRY_SIZE - (msg_start - (char *)entry);
 259         utf8_to_ascii(msg_start, buf_len, msg, TRUE);
 260
 261         uint32_t msg_len = strlen(msg_start) + 1;
 262
 263         uint32_t size = sizeof(clog_entry_t) + node_len + ident_len +
 264                 tag_len + msg_len;
 265
 266         if (size > CLOG_MAX_ENTRY_SIZE) {
 267                 int diff = size - CLOG_MAX_ENTRY_SIZE;
 268                 msg_len -= diff;
 269                 size = CLOG_MAX_ENTRY_SIZE;
 270         }
 271
 272         entry->prev = 0;
 273         entry->next = 0;
 274         entry->uid = ++uid_counter;
 275         entry->time = logtime;
 276         entry->node_digest = fnv_64a_buf(node, node_len, FNV1A_64_INIT);
 277         entry->ident_digest = fnv_64a_buf(ident, ident_len, FNV1A_64_INIT);
 278         entry->pid = pid;
 279         entry->priority = priority;
 280         entry->node_len = node_len;
 281         entry->ident_len = ident_len;
 282         entry->tag_len = tag_len;
 283         entry->msg_len = msg_len;
 284
 285         char *p = entry->data;
 286         g_strlcpy(p, node, node_len);
 287         p = p + node_len;
 288         g_strlcpy(p, ident, ident_len);
 289         p = p + ident_len;
 290         g_strlcpy(p, tag, tag_len);
 291
 292         return size;
 293 }
 294
 295 clog_base_t *
 296 clog_new(uint32_t size)
 297 {
 298         g_return_val_if_fail(sizeof(clog_base_t) == 8, NULL);
 299
 300         if (!size)
 301                 size = CLOG_DEFAULT_SIZE;
 302
 303         g_return_val_if_fail(size >= (CLOG_MAX_ENTRY_SIZE*10), NULL);
 304
 305
 306         clog_base_t *clog = (clog_base_t *)g_malloc0(size);
 307         if (clog) {
 308                 clog->size = size;
 309         }
 310
 311         return clog;
 312 }
 313
 314 static gint
 315 clog_entry_sort_fn(
 316         gconstpointer v1,
 317         gconstpointer v2,
 318         gpointer user_data)
 319 {
 320         clog_entry_t *entry1 = (clog_entry_t *)v1;
 321         clog_entry_t *entry2 = (clog_entry_t *)v2;
 322
 323         if (entry1->time != entry2->time)
 324                 return entry1->time - entry2->time;
 325
 326         if (entry1->node_digest != entry2->node_digest)
 327                 return entry1->node_digest - entry2->node_digest;
 328
 329         return entry1->uid - entry2->uid;
 330 }
 331
 332 static gboolean
 333 clog_tree_foreach_fn(
 334         gpointer key,
 335         gpointer value,
 336         gpointer data)
 337 {
 338         clog_entry_t *entry = (clog_entry_t *)value;
 339         clog_base_t *clog = (clog_base_t *)data;
 340
 341         clog_copy(clog, entry);
 342
 343         return FALSE;
 344 }
 345
 346 clog_base_t *
 347 clog_sort(clog_base_t *clog)
 348 {
 349         g_return_val_if_fail(clog != NULL, NULL);
 350         g_return_val_if_fail(clog->cpos != 0, NULL);
 351
 352         clog_base_t *res = clog_new(clog->size);
 353         if (!res)
 354                 return NULL;
 355
 356         GTree *tree = g_tree_new_with_data(clog_entry_sort_fn, NULL);
 357         if (!tree) {
 358                 g_free(res);
 359                 return NULL;
 360         }
 361
 362         uint32_t cpos = clog->cpos;
 363
 364         while (cpos && (cpos <= clog->cpos || cpos > (clog->cpos + CLOG_MAX_ENTRY_SIZE))) {
 365                 if (cpos > (clog->size - sizeof(clog_entry_t))) {
 366                         cfs_critical("log pointer out of range!");
 367                         break;
 368                 }
 369                 clog_entry_t *cur = (clog_entry_t *)((char *)clog + cpos);
 370
 371                 g_tree_insert(tree, cur, cur);
 372
 373                 cpos = cur->prev;
 374         }
 375
 376         g_tree_foreach(tree, clog_tree_foreach_fn, res);
 377         g_tree_destroy(tree);
 378
 379         return res;
 380 }
 381
 382 uint32_t
 383 clog_size(clog_base_t *clog)
 384 {
 385         g_return_val_if_fail(clog != NULL, 0);
 386         return clog->size;
 387 }
 388
 389 static gboolean
 390 dedup_lookup(
 391         GHashTable *dedup,
 392         const clog_entry_t *entry)
 393 {
 394         g_return_val_if_fail(dedup != NULL, FALSE);
 395         g_return_val_if_fail(entry != NULL, FALSE);
 396
 397         dedup_entry_t *dd = g_hash_table_lookup(dedup, &entry->node_digest);
 398         if (!dd) {
 399                 if (!(dd = g_new0(dedup_entry_t, 1)))
 400                         return FALSE;
 401
 402                 dd->node_digest = entry->node_digest;
 403                 dd->time = entry->time;
 404                 dd->uid = entry->uid;
 405
 406                 g_hash_table_insert(dedup, dd, dd);
 407
 408                 return TRUE;
 409         }
 410
 411         if (entry->time > dd->time ||
 412             (entry->time == dd->time && entry->uid > dd->uid)) {
 413                 dd->time = entry->time;
 414                 dd->uid = entry->uid;
 415                 return TRUE;
 416         }
 417
 418         return FALSE;
 419 }
 420
 421 struct clusterlog {
 422         GHashTable *dedup;
 423         GMutex *mutex;
 424         clog_base_t *base;
 425 };
 426
 427 void
 428 clusterlog_dump(
 429         clusterlog_t *cl,
 430         GString *str,
 431         const char *user,
 432         guint max_entries)
 433 {
 434         g_return_if_fail(cl != NULL);
 435         g_return_if_fail(str != NULL);
 436
 437         g_mutex_lock(cl->mutex);
 438         clog_dump_json(cl->base, str, user, max_entries);
 439         g_mutex_unlock(cl->mutex);
 440 }
 441
 442 clog_base_t *
 443 clusterlog_merge(
 444         clusterlog_t *cl,
 445         clog_base_t **clog,
 446         int count,
 447         int local_index)
 448 {
 449         g_return_val_if_fail(cl != NULL, NULL);
 450         g_return_val_if_fail(clog != NULL, NULL);
 451         g_return_val_if_fail(count >= 2, NULL);
 452         g_return_val_if_fail(local_index >= 0, NULL);
 453         g_return_val_if_fail(local_index < count, NULL);
 454
 455         uint32_t cpos[count];
 456         uint32_t maxsize = 0;
 457
 458         GHashTable *dedup;
 459         if (!(dedup = g_hash_table_new_full(g_int64_hash, g_int64_equal, NULL, g_free)))
 460                 return NULL;
 461
 462         GTree *tree = g_tree_new_with_data(clog_entry_sort_fn, NULL);
 463         if (!tree) {
 464                 g_hash_table_destroy(dedup);
 465                 return NULL;
 466         }
 467
 468         clog_base_t *res = clog_new(maxsize);
 469         if (!res) {
 470                 g_hash_table_destroy(dedup);
 471                 g_tree_destroy(tree);
 472                 return NULL;
 473         }
 474
 475         g_mutex_lock(cl->mutex);
 476
 477         for (int i = 0; i < count; i++) {
 478                 if (i == local_index)
 479                         clog[i] = cl->base;
 480
 481                 if (!clog[i]) {
 482                         cfs_critical("log pointer is NULL!");
 483                         cpos[i] = 0;
 484                         continue;
 485                 }
 486                 cpos[i] = clog[i]->cpos;
 487                 if (clog[i]->size > maxsize)
 488                         maxsize = clog[i]->size;
 489         }
 490
 491         size_t logsize = 0;
 492         maxsize = res->size - sizeof(clog_base_t) - CLOG_MAX_ENTRY_SIZE;
 493
 494         int found = 0;
 495         while (found >= 0) {
 496
 497                 found = -1;
 498                 uint32_t last = 0;
 499
 500                 /* select entry wit latest time */
 501                 for (int i = 0; i < count; i++) {
 502                         if (!cpos[i])
 503                                 continue;
 504                         clog_entry_t *cur = (clog_entry_t *)((char *)clog[i] + cpos[i]);
 505                         if (cur->time > last) {
 506                                 last = cur->time;
 507                                 found = i;
 508                         }
 509                 }
 510
 511                 if (found < 0)
 512                         break;
 513
 514                 clog_entry_t *cur = (clog_entry_t *)((char *)clog[found] + cpos[found]);
 515
 516                 if (!g_tree_lookup(tree, cur)) {
 517                         g_tree_insert(tree, cur, cur);
 518                         dedup_lookup(dedup, cur); /* just to record versions */
 519                         logsize += cur->next - cpos[found];
 520                         if (logsize >= maxsize)
 521                                 break;
 522                 }
 523
 524                 if (!cur->prev) {
 525                         cpos[found] = 0;
 526                 } else {
 527                         cpos[found] = cur->prev;
 528                         if (!(cpos[found] <= clog[found]->cpos ||
 529                               cpos[found] > (clog[found]->cpos + CLOG_MAX_ENTRY_SIZE))) {
 530                                 cpos[found] = 0;
 531                         }
 532                 }
 533         }
 534
 535         g_tree_foreach(tree, clog_tree_foreach_fn, res);
 536         g_tree_destroy(tree);
 537
 538         g_hash_table_destroy(cl->dedup);
 539         cl->dedup = dedup;
 540
 541         g_free(cl->base);
 542         cl->base = res;
 543
 544         g_mutex_unlock(cl->mutex);
 545
 546         return res;
 547 }
 548
 549 void
 550 clusterlog_destroy(clusterlog_t *cl)
 551 {
 552         g_return_if_fail(cl != NULL);
 553
 554         if (cl->mutex)
 555                 g_mutex_free(cl->mutex);
 556
 557         if (cl->base)
 558                 g_free(cl->base);
 559
 560         if (cl->dedup)
 561                 g_hash_table_destroy(cl->dedup);
 562
 563         g_free(cl);
 564 }
 565
 566 clusterlog_t *
 567 clusterlog_new(void)
 568 {
 569         clusterlog_t *cl = g_new0(clusterlog_t, 1);
 570         if (!cl)
 571                 return NULL;
 572
 573         if (!(cl->mutex = g_mutex_new()))
 574                 goto fail;
 575
 576         if (!(cl->base = clog_new(0)))
 577                 goto fail;
 578
 579         if (!(cl->dedup = g_hash_table_new_full(g_int64_hash, g_int64_equal, NULL, g_free)))
 580                 goto fail;
 581
 582         return cl;
 583
 584 fail:
 585         clusterlog_destroy(cl);
 586         return NULL;
 587 }
 588
 589 gpointer
 590 clusterlog_get_state(
 591         clusterlog_t *cl,
 592         unsigned int *res_len)
 593 {
 594         g_return_val_if_fail(cl != NULL, NULL);
 595         g_return_val_if_fail(res_len != NULL, NULL);
 596
 597         g_mutex_lock(cl->mutex);
 598
 599         clog_base_t *new;
 600         if ((new = clog_sort(cl->base))) {
 601                 g_free(cl->base);
 602                 cl->base = new;
 603         }
 604
 605         *res_len = clog_size(cl->base);
 606         gpointer msg = g_memdup(cl->base, *res_len);
 607
 608         g_mutex_unlock(cl->mutex);
 609
 610         return msg;
 611 }
 612
 613 void
 614 clusterlog_insert(
 615         clusterlog_t *cl,
 616         const clog_entry_t *entry)
 617 {
 618         g_return_if_fail(cl != NULL);
 619         g_return_if_fail(entry != NULL);
 620
 621         g_mutex_lock(cl->mutex);
 622
 623         if (dedup_lookup(cl->dedup, entry)) {
 624                 clog_copy(cl->base, entry);
 625         } else {
 626                 cfs_message("ignore duplicate"); // fixme remove
 627         }
 628
 629         g_mutex_unlock(cl->mutex);
 630 };
 631
 632 void
 633 clusterlog_add(
 634         clusterlog_t *cl,
 635         const char *ident,
 636         const char *tag,
 637         uint32_t pid,
 638         uint8_t priority,
 639         const gchar *format,
 640         ...)
 641 {
 642         g_return_if_fail(cl != NULL);
 643         g_return_if_fail(format != NULL);
 644
 645         va_list args;
 646         va_start (args, format);
 647         char *msg = g_strdup_vprintf (format, args);
 648         va_end (args);
 649
 650         time_t ctime = time(NULL);
 651         clog_entry_t *entry = (clog_entry_t *)alloca(CLOG_MAX_ENTRY_SIZE);
 652         uint32_t size = clog_pack(entry, cfs.nodename, ident, tag, pid, ctime, priority, msg);
 653         g_free(msg);
 654
 655         if (!size)
 656                 return;
 657
 658         clusterlog_insert(cl, entry);
 659 }
 660