1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
|
// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
#include "protocol.h"
#include <lz4.h>
#include <vespa/vespalib/util/stringfmt.h>
#include <vespa/vespalib/data/slime/slime.h>
#include <sstream>
#include <cassert>
#include <vespa/log/log.h>
LOG_SETUP(".config.frt.protocol");
using namespace vespalib;
using vespalib::alloc::Alloc;
using namespace vespalib::slime;
namespace config {
namespace protocol {
namespace v2 {
const Memory REQUEST_VERSION = "version";
const Memory REQUEST_DEF_NAME = "defName";
const Memory REQUEST_DEF_NAMESPACE = "defNamespace";
const Memory REQUEST_DEF_MD5 = "defMD5";
const Memory REQUEST_DEF_CONTENT = "defContent";
const Memory REQUEST_CLIENT_CONFIGID = "configId";
const Memory REQUEST_CLIENT_HOSTNAME = "clientHostname";
const Memory REQUEST_CONFIG_MD5 = "configMD5";
const Memory REQUEST_CONFIG_XXHASH64 = "configXxhash64";
const Memory REQUEST_CURRENT_GENERATION = "currentGeneration";
const Memory REQUEST_TIMEOUT = "timeout";
const Memory REQUEST_TRACE = "trace";
const Memory REQUEST_VESPA_VERSION = "vespaVersion";
const Memory RESPONSE_VERSION = "version";
const Memory RESPONSE_DEF_NAME = "defName";
const Memory RESPONSE_DEF_NAMESPACE = "defNamespace";
const Memory RESPONSE_DEF_MD5 = "defMD5";
const Memory RESPONSE_CONFIGID = "configId";
const Memory RESPONSE_CLIENT_HOSTNAME = "clientHostname";
const Memory RESPONSE_CONFIG_XXHASH64 = "configXxhash64";
const Memory RESPONSE_CONFIG_GENERATION = "generation";
const Memory RESPONSE_PAYLOAD = "payload";
const Memory RESPONSE_TRACE = "trace";
const Memory RESPONSE_APPLY_ON_RESTART = "applyOnRestart";
const Inspector &
extractPayload(const Slime & data)
{
const Inspector & payload(data.get()[RESPONSE_PAYLOAD]);
if (LOG_WOULD_LOG(debug)) {
LOG(debug, "payload: %s", payload.toString().c_str());
}
return payload;
}
}
namespace v3 {
const Memory REQUEST_COMPRESSION_TYPE = "compressionType";
const Memory RESPONSE_COMPRESSION_INFO = "compressionInfo";
const Memory RESPONSE_COMPRESSION_INFO_TYPE = "compressionType";
const Memory RESPONSE_COMPRESSION_INFO_UNCOMPRESSED_SIZE = "uncompressedSize";
DecompressedData
decompress_lz4(const char * input, uint32_t inputLen, int uncompressedLength)
{
Alloc memory( Alloc::alloc(uncompressedLength));
int sz = LZ4_decompress_safe(input, static_cast<char *>(memory.get()), inputLen, uncompressedLength);
if (sz >= 0 && sz != uncompressedLength) {
if (LOG_WOULD_LOG(debug)) {
LOG(debug, "Returned compressed size (%d) is not the same as uncompressed size(%d)", sz, uncompressedLength);
}
Alloc copy = memory.create(sz);
memcpy(copy.get(), memory.get(), sz);
memory = std::move(copy);
}
assert(sz >= 0);
return DecompressedData(std::move(memory), static_cast<uint32_t>(sz));
}
DecompressedData
decompress(const char * input, uint32_t len, const CompressionType & compressionType, uint32_t uncompressedLength)
{
// No payload means no data
if (len == 0) {
return DecompressedData(Memory(input, len), len);
}
switch (compressionType) {
case CompressionType::LZ4:
return decompress_lz4(input, len, uncompressedLength);
break;
case CompressionType::UNCOMPRESSED:
default:
return DecompressedData(Memory(input, len), len);
break;
}
}
}
const int DEFAULT_PROTOCOL_VERSION = 3;
const int DEFAULT_TRACE_LEVEL = 0;
int
verifyProtocolVersion(int protocolVersion)
{
if (1 == protocolVersion || 2 == protocolVersion || 3 == protocolVersion) {
return protocolVersion;
}
LOG(info, "Unknown protocol version %d, using default (%d)", protocolVersion, DEFAULT_PROTOCOL_VERSION);
return DEFAULT_PROTOCOL_VERSION;
}
int
readProtocolVersion()
{
int protocolVersion = DEFAULT_PROTOCOL_VERSION;
char *versionStringPtr = getenv("VESPA_CONFIG_PROTOCOL_VERSION");
if (versionStringPtr != NULL) {
std::stringstream versionString(versionStringPtr);
versionString >> protocolVersion;
}
return verifyProtocolVersion(protocolVersion);
}
int
readTraceLevel()
{
int traceLevel = DEFAULT_TRACE_LEVEL;
char *traceLevelStringPtr = getenv("VESPA_CONFIG_PROTOCOL_TRACELEVEL");
if (traceLevelStringPtr != NULL) {
std::stringstream traceLevelString(traceLevelStringPtr);
traceLevelString >> traceLevel;
}
return traceLevel;
}
CompressionType
readProtocolCompressionType()
{
CompressionType type = CompressionType::LZ4;
char *compressionTypeStringPtr = getenv("VESPA_CONFIG_PROTOCOL_COMPRESSION");
if (compressionTypeStringPtr != NULL) {
type = stringToCompressionType(vespalib::string(compressionTypeStringPtr));
}
return type;
}
}
}
|
