@cryptotaxi247 / netdata-1 / commits / f1c26d0e2

DBENGINE: support ZSTD compression (#17244)

* extract dbengine compression to separate files * added ZSTD support in dbengine * automatically select best compression * handle decompression errors * eliminate fatals from compression algorithms; fallback to uncompressed pages if compression fails or generates bigger data * have the unit test generate many data files

Costa Tsaousis committed Mar 25, 2024 at 12:30 UTC f1c26d0e2bf2f09480b48a5e9f9dc7abea88a5da

7 files changed +223 -41

CMakeLists.txt

@@ -1054,6 +1054,8 @@ if(ENABLE_DBENGINE)

1054 src/database/engine/pdc.h

1055 src/database/engine/dbengine-unittest.c

1056 src/database/engine/dbengine-stresstest.c

1057 + src/database/engine/dbengine-compression.c

1058 + src/database/engine/dbengine-compression.h

1059 )

1060 endif()

1061

src/database/engine/dbengine-compression.c new

+163

@@ -0,0 +1,163 @@

1 +// SPDX-License-Identifier: GPL-3.0-or-later

2 +

3 +#include "rrdengine.h"

4 +#include "dbengine-compression.h"

5 +

6 +#ifdef ENABLE_LZ4

7 +#include <lz4.h>

8 +#endif

9 +

10 +#ifdef ENABLE_ZSTD

11 +#include <zstd.h>

12 +#define DBENGINE_ZSTD_DEFAULT_COMPRESSION_LEVEL 3

13 +#endif

14 +

15 +uint8_t dbengine_default_compression(void) {

16 +

17 +#ifdef ENABLE_ZSTD

18 + return RRDENG_COMPRESSION_ZSTD;

19 +#endif

20 +

21 +#ifdef ENABLE_LZ4

22 + return RRDENG_COMPRESSION_LZ4;

23 +#endif

24 +

25 + return RRDENG_COMPRESSION_NONE;

26 +}

27 +

28 +bool dbengine_valid_compression_algorithm(uint8_t algorithm) {

29 + switch(algorithm) {

30 + case RRDENG_COMPRESSION_NONE:

31 +

32 +#ifdef ENABLE_LZ4

33 + case RRDENG_COMPRESSION_LZ4:

34 +#endif

35 +

36 +#ifdef ENABLE_ZSTD

37 + case RRDENG_COMPRESSION_ZSTD:

38 +#endif

39 +

40 + return true;

41 +

42 + default:

43 + return false;

44 + }

45 +}

46 +

47 +size_t dbengine_max_compressed_size(size_t uncompressed_size, uint8_t algorithm) {

48 + switch(algorithm) {

49 +#ifdef ENABLE_LZ4

50 + case RRDENG_COMPRESSION_LZ4:

51 + fatal_assert(uncompressed_size < LZ4_MAX_INPUT_SIZE);

52 + return LZ4_compressBound((int)uncompressed_size);

53 +#endif

54 +

55 +#ifdef ENABLE_ZSTD

56 + case RRDENG_COMPRESSION_ZSTD:

57 + return ZSTD_compressBound(uncompressed_size);

58 +#endif

59 +

60 + case RRDENG_COMPRESSION_NONE:

61 + return uncompressed_size;

62 +

63 + default:

64 + fatal("DBENGINE: unknown compression algorithm %u", algorithm);

65 + }

66 +}

67 +

68 +size_t dbengine_compress(void *payload, size_t uncompressed_size, uint8_t algorithm) {

69 + // the result should be stored in the payload

70 + // the caller must have called dbengine_max_compressed_size() to make sure the

71 + // payload is big enough to fit the max size needed.

72 +

73 + switch(algorithm) {

74 +#ifdef ENABLE_LZ4

75 + case RRDENG_COMPRESSION_LZ4: {

76 + size_t max_compressed_size = dbengine_max_compressed_size(uncompressed_size, algorithm);

77 + struct extent_buffer *eb = extent_buffer_get(max_compressed_size);

78 + void *compressed_buf = eb->data;

79 +

80 + size_t compressed_size =

81 + LZ4_compress_default(payload, compressed_buf, (int)uncompressed_size, (int)max_compressed_size);

82 +

83 + if(compressed_size > 0 && compressed_size < uncompressed_size)

84 + memcpy(payload, compressed_buf, compressed_size);

85 + else

86 + compressed_size = 0;

87 +

88 + extent_buffer_release(eb);

89 + return compressed_size;

90 + }

91 +#endif

92 +

93 +#ifdef ENABLE_ZSTD

94 + case RRDENG_COMPRESSION_ZSTD: {

95 + size_t max_compressed_size = dbengine_max_compressed_size(uncompressed_size, algorithm);

96 + struct extent_buffer *eb = extent_buffer_get(max_compressed_size);

97 + void *compressed_buf = eb->data;

98 +

99 + size_t compressed_size = ZSTD_compress(compressed_buf, max_compressed_size, payload, uncompressed_size,

100 + DBENGINE_ZSTD_DEFAULT_COMPRESSION_LEVEL);

101 +

102 + if (ZSTD_isError(compressed_size)) {

103 + internal_fatal(true, "DBENGINE: ZSTD compression error %s", ZSTD_getErrorName(compressed_size));

104 + compressed_size = 0;

105 + }

106 +

107 + if(compressed_size > 0 && compressed_size < uncompressed_size)

108 + memcpy(payload, compressed_buf, compressed_size);

109 + else

110 + compressed_size = 0;

111 +

112 + extent_buffer_release(eb);

113 + return compressed_size;

114 + }

115 +#endif

116 +

117 + case RRDENG_COMPRESSION_NONE:

118 + return 0;

119 +

120 + default:

121 + fatal("DBENGINE: unknown compression algorithm %u", algorithm);

122 + }

123 +}

124 +

125 +size_t dbengine_decompress(void *dst, void *src, size_t dst_size, size_t src_size, uint8_t algorithm) {

126 + switch(algorithm) {

127 +

128 +#ifdef ENABLE_LZ4

129 + case RRDENG_COMPRESSION_LZ4: {

130 + int rc = LZ4_decompress_safe(src, dst, (int)src_size, (int)dst_size);

131 + if(rc < 0) {

132 + nd_log(NDLS_DAEMON, NDLP_ERR, "DBENGINE: ZSTD decompression error %d", rc);

133 + rc = 0;

134 + }

135 +

136 + return rc;

137 + }

138 +#endif

139 +

140 +#ifdef ENABLE_ZSTD

141 + case RRDENG_COMPRESSION_ZSTD: {

142 + size_t decompressed_size = ZSTD_decompress(dst, dst_size, src, src_size);

143 +

144 + if (ZSTD_isError(decompressed_size)) {

145 + nd_log(NDLS_DAEMON, NDLP_ERR, "DBENGINE: ZSTD decompression error %s",

146 + ZSTD_getErrorName(decompressed_size));

147 +

148 + decompressed_size = 0;

149 + }

150 +

151 + return decompressed_size;

152 + }

153 +#endif

154 +

155 + case RRDENG_COMPRESSION_NONE:

156 + internal_fatal(true, "DBENGINE: %s() should not be called for uncompressed pages", __FUNCTION__ );

157 + return 0;

158 +

159 + default:

160 + internal_fatal(true, "DBENGINE: unknown compression algorithm %u", algorithm);

161 + return 0;

162 + }

163 +}

src/database/engine/dbengine-compression.h new

+15

@@ -0,0 +1,15 @@

1 +// SPDX-License-Identifier: GPL-3.0-or-later

2 +

3 +#ifndef NETDATA_DBENGINE_COMPRESSION_H

4 +#define NETDATA_DBENGINE_COMPRESSION_H

5 +

6 +uint8_t dbengine_default_compression(void);

7 +

8 +bool dbengine_valid_compression_algorithm(uint8_t algorithm);

9 +

10 +size_t dbengine_max_compressed_size(size_t uncompressed_size, uint8_t algorithm);

11 +size_t dbengine_compress(void *payload, size_t uncompressed_size, uint8_t algorithm);

12 +

13 +size_t dbengine_decompress(void *dst, void *src, size_t dst_size, size_t src_size, uint8_t algorithm);

14 +

15 +#endif //NETDATA_DBENGINE_COMPRESSION_H

src/database/engine/pdc.c

+12 -8

@@ -1,6 +1,7 @@

1 // SPDX-License-Identifier: GPL-3.0-or-later

2 #define NETDATA_RRD_INTERNALS

3 #include "pdc.h"

4 +#include "dbengine-compression.h"

6 struct extent_page_details_list {

7 uv_file file;

@@ -940,7 +941,6 @@ static bool epdl_populate_pages_from_extent_data(

941 PDC_PAGE_STATUS tags,

942 bool cached_extent)

943 {

943 - int ret;

944 unsigned i, count;

945 void *uncompressed_buf = NULL;

946 uint32_t payload_length, payload_offset, trailer_offset, uncompressed_payload_length = 0;

@@ -975,7 +975,7 @@ static bool epdl_populate_pages_from_extent_data(

975 if( !can_use_data ||

976 count < 1 ||

977 count > MAX_PAGES_PER_EXTENT ||

978 - (header->compression_algorithm != RRDENG_COMPRESSION_NONE && header->compression_algorithm != RRDENG_COMPRESSION_LZ4) ||

978 + !dbengine_valid_compression_algorithm(header->compression_algorithm) ||

979 (payload_length != trailer_offset - payload_offset) ||

980 (data_length != payload_offset + payload_length + sizeof(*trailer))

981 ) {

@@ -985,8 +985,7 @@ static bool epdl_populate_pages_from_extent_data(

985

986 crc = crc32(0L, Z_NULL, 0);

987 crc = crc32(crc, data, epdl->extent_size - sizeof(*trailer));

988 - ret = crc32cmp(trailer->checksum, crc);

989 - if (unlikely(ret)) {

988 + if (unlikely(crc32cmp(trailer->checksum, crc))) {

989 ctx_io_error(ctx);

990 have_read_error = true;

991 epdl_extent_loading_error_log(ctx, epdl, NULL, "CRC32 checksum FAILED");

@@ -1018,11 +1017,16 @@ static bool epdl_populate_pages_from_extent_data(

1017 eb = extent_buffer_get(uncompressed_payload_length);

1018 uncompressed_buf = eb->data;

1019

1021 - ret = LZ4_decompress_safe(data + payload_offset, uncompressed_buf,

1022 - (int) payload_length, (int) uncompressed_payload_length);

1020 + size_t bytes = dbengine_decompress(uncompressed_buf, data + payload_offset,

1021 + uncompressed_payload_length, payload_length,

1022 + header->compression_algorithm);

1023

1024 - __atomic_add_fetch(&ctx->stats.before_decompress_bytes, payload_length, __ATOMIC_RELAXED);

1025 - __atomic_add_fetch(&ctx->stats.after_decompress_bytes, ret, __ATOMIC_RELAXED);

1024 + if(!bytes)

1025 + have_read_error = true;

1026 + else {

1027 + __atomic_add_fetch(&ctx->stats.before_decompress_bytes, payload_length, __ATOMIC_RELAXED);

1028 + __atomic_add_fetch(&ctx->stats.after_decompress_bytes, bytes, __ATOMIC_RELAXED);

1029 + }

1030 }

1031 }

1032

src/database/engine/rrddiskprotocol.h

@@ -21,6 +21,7 @@

22 #define RRDENG_COMPRESSION_NONE (0)

23 #define RRDENG_COMPRESSION_LZ4 (1)

24 +#define RRDENG_COMPRESSION_ZSTD (2)

26 #define RRDENG_DF_SB_PADDING_SZ (RRDENG_BLOCK_SIZE - (RRDENG_MAGIC_SZ + RRDENG_VER_SZ + sizeof(uint8_t)))

src/database/engine/rrdengine.c

+28 -32

@@ -3,6 +3,7 @@

4 #include "rrdengine.h"

5 #include "pdc.h"

6 +#include "dbengine-compression.h"

8 rrdeng_stats_t global_io_errors = 0;

9 rrdeng_stats_t global_fs_errors = 0;

@@ -772,13 +773,10 @@ static struct rrdengine_datafile *get_datafile_to_write_extent(struct rrdengine_

773 */

774 static struct extent_io_descriptor *datafile_extent_build(struct rrdengine_instance *ctx, struct page_descr_with_data *base, struct completion *completion) {

775 int ret;

775 - int compressed_size, max_compressed_size = 0;

776 unsigned i, count, size_bytes, pos, real_io_size;

777 - uint32_t uncompressed_payload_length, payload_offset;

777 + uint32_t uncompressed_payload_length, max_compressed_size, payload_offset;

778 struct page_descr_with_data *descr, *eligible_pages[MAX_PAGES_PER_EXTENT];

779 struct extent_io_descriptor *xt_io_descr;

780 - struct extent_buffer *eb = NULL;

781 - void *compressed_buf = NULL;

780 Word_t Index;

781 uint8_t compression_algorithm = ctx->config.global_compress_alg;

782 struct rrdengine_datafile *datafile;

@@ -807,20 +805,8 @@ static struct extent_io_descriptor *datafile_extent_build(struct rrdengine_insta

805 xt_io_descr = extent_io_descriptor_get();

806 xt_io_descr->ctx = ctx;

807 payload_offset = sizeof(*header) + count * sizeof(header->descr[0]);

810 - switch (compression_algorithm) {

811 - case RRDENG_COMPRESSION_NONE:

812 - size_bytes = payload_offset + uncompressed_payload_length + sizeof(*trailer);

813 - break;

814 -

815 - default: /* Compress */

816 - fatal_assert(uncompressed_payload_length < LZ4_MAX_INPUT_SIZE);

817 - max_compressed_size = LZ4_compressBound(uncompressed_payload_length);

818 - eb = extent_buffer_get(max_compressed_size);

819 - compressed_buf = eb->data;

820 - size_bytes = payload_offset + MAX(uncompressed_payload_length, (unsigned)max_compressed_size) + sizeof(*trailer);

821 - break;

822 - }

823 -

808 + max_compressed_size = dbengine_max_compressed_size(uncompressed_payload_length, compression_algorithm);

809 + size_bytes = payload_offset + MAX(uncompressed_payload_length, max_compressed_size) + sizeof(*trailer);

810 ret = posix_memalign((void *)&xt_io_descr->buf, RRDFILE_ALIGNMENT, ALIGN_BYTES_CEILING(size_bytes));

811 if (unlikely(ret)) {

812 fatal("DBENGINE: posix_memalign:%s", strerror(ret));

@@ -832,7 +818,6 @@ static struct extent_io_descriptor *datafile_extent_build(struct rrdengine_insta

818

819 pos = 0;

820 header = xt_io_descr->buf;

835 - header->compression_algorithm = compression_algorithm;

821 header->number_of_pages = count;

822 pos += sizeof(*header);

823

@@ -858,29 +843,40 @@ static struct extent_io_descriptor *datafile_extent_build(struct rrdengine_insta

843

844 pos += sizeof(header->descr[i]);

845 }

846 +

847 + // build the extent payload

848 for (i = 0 ; i < count ; ++i) {

849 descr = xt_io_descr->descr_array[i];

850 pgd_copy_to_extent(descr->pgd, xt_io_descr->buf + pos, descr->page_length);

851 pos += descr->page_length;

852 }

853

867 - if(likely(compression_algorithm == RRDENG_COMPRESSION_LZ4)) {

868 - compressed_size = LZ4_compress_default(

869 - xt_io_descr->buf + payload_offset,

870 - compressed_buf,

871 - (int)uncompressed_payload_length,

872 - max_compressed_size);

854 + // compress the payload

855 + size_t compressed_size =

856 + (int)dbengine_compress(xt_io_descr->buf + payload_offset,

857 + uncompressed_payload_length,

858 + compression_algorithm);

859

874 - __atomic_add_fetch(&ctx->stats.before_compress_bytes, uncompressed_payload_length, __ATOMIC_RELAXED);

875 - __atomic_add_fetch(&ctx->stats.after_compress_bytes, compressed_size, __ATOMIC_RELAXED);

860 + internal_fatal(compressed_size > max_compressed_size, "DBENGINE: compression returned more data than the max allowed");

861 + internal_fatal(compressed_size > uncompressed_payload_length, "DBENGINE: compression returned more data than the uncompressed extent");

862

877 - (void) memcpy(xt_io_descr->buf + payload_offset, compressed_buf, compressed_size);

878 - extent_buffer_release(eb);

879 - size_bytes = payload_offset + compressed_size + sizeof(*trailer);

863 + if(compressed_size) {

864 + header->compression_algorithm = compression_algorithm;

865 header->payload_length = compressed_size;

866 }

882 - else { // RRD_NO_COMPRESSION

883 - header->payload_length = uncompressed_payload_length;

867 + else {

868 + // compression failed, or generated bigger pages

869 + // so it didn't touch our uncompressed buffer

870 + header->compression_algorithm = RRDENG_COMPRESSION_NONE;

871 + header->payload_length = compressed_size = uncompressed_payload_length;

872 + }

873 +

874 + // set the correct size

875 + size_bytes = payload_offset + compressed_size + sizeof(*trailer);

876 +

877 + if(compression_algorithm != RRDENG_COMPRESSION_NONE) {

878 + __atomic_add_fetch(&ctx->stats.before_compress_bytes, uncompressed_payload_length, __ATOMIC_RELAXED);

879 + __atomic_add_fetch(&ctx->stats.after_compress_bytes, compressed_size, __ATOMIC_RELAXED);

880 }

881

882 real_io_size = ALIGN_BYTES_CEILING(size_bytes);

src/database/engine/rrdengineapi.c

+2 -1

@@ -2,6 +2,7 @@

3 #include "database/engine/rrddiskprotocol.h"

4 #include "rrdengine.h"

5 +#include "dbengine-compression.h"

7 /* Default global database instance */

8 struct rrdengine_instance multidb_ctx_storage_tier0;

@@ -1170,7 +1171,7 @@ int rrdeng_init(struct rrdengine_instance **ctxp, const char *dbfiles_path,

1171

1172 ctx->config.tier = (int)tier;

1173 ctx->config.page_type = tier_page_type[tier];

1173 - ctx->config.global_compress_alg = RRDENG_COMPRESSION_LZ4;

1174 + ctx->config.global_compress_alg = dbengine_default_compression();

1175 if (disk_space_mb < RRDENG_MIN_DISK_SPACE_MB)

1176 disk_space_mb = RRDENG_MIN_DISK_SPACE_MB;

1177 ctx->config.max_disk_space = disk_space_mb * 1048576LLU;