summaryrefslogtreecommitdiff
path: root/src/mongo/s/client/version_manager.cpp
blob: f1cdfbc5e6b8ec594876aab1c72d15ec83c6f9c4 (plain)
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
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
/**
 *    Copyright (C) 2018-present MongoDB, Inc.
 *
 *    This program is free software: you can redistribute it and/or modify
 *    it under the terms of the Server Side Public License, version 1,
 *    as published by MongoDB, Inc.
 *
 *    This program is distributed in the hope that it will be useful,
 *    but WITHOUT ANY WARRANTY; without even the implied warranty of
 *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *    Server Side Public License for more details.
 *
 *    You should have received a copy of the Server Side Public License
 *    along with this program. If not, see
 *    <http://www.mongodb.com/licensing/server-side-public-license>.
 *
 *    As a special exception, the copyright holders give permission to link the
 *    code of portions of this program with the OpenSSL library under certain
 *    conditions as described in each individual source file and distribute
 *    linked combinations including the program with the OpenSSL library. You
 *    must comply with the Server Side Public License in all respects for
 *    all of the code used other than as permitted herein. If you modify file(s)
 *    with this exception, you may extend this exception to your version of the
 *    file(s), but you are not obligated to do so. If you do not wish to do so,
 *    delete this exception statement from your version. If you delete this
 *    exception statement from all source files in the program, then also delete
 *    it in the license file.
 */

#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kSharding

#include "mongo/platform/basic.h"

#include "mongo/s/client/version_manager.h"

#include "mongo/client/dbclient_cursor.h"
#include "mongo/client/dbclient_rs.h"
#include "mongo/db/namespace_string.h"
#include "mongo/s/catalog_cache.h"
#include "mongo/s/client/shard_connection.h"
#include "mongo/s/client/shard_registry.h"
#include "mongo/s/grid.h"
#include "mongo/s/is_mongos.h"
#include "mongo/s/request_types/set_shard_version_request.h"
#include "mongo/s/stale_exception.h"
#include "mongo/s/transaction_router.h"
#include "mongo/util/log.h"

namespace mongo {

using std::shared_ptr;
using std::map;
using std::string;

namespace {

/**
 * Tracking information, per-connection, of the latest chunk manager iteration or sequence
 * number that was used to send a shard version over this connection.
 * When the chunk manager is replaced, implying new versions were loaded, the chunk manager
 * sequence number is iterated by 1 and connections need to re-send shard versions.
 */
class ConnectionShardStatus {
public:
    bool hasAnySequenceSet(DBClientBase* conn) {
        stdx::lock_guard<stdx::mutex> lk(_mutex);

        SequenceMap::const_iterator seenConnIt = _map.find(conn->getConnectionId());
        return seenConnIt != _map.end() && seenConnIt->second.size() > 0;
    }

    bool getSequence(DBClientBase* conn, const string& ns, unsigned long long* sequence) {
        stdx::lock_guard<stdx::mutex> lk(_mutex);

        SequenceMap::const_iterator seenConnIt = _map.find(conn->getConnectionId());
        if (seenConnIt == _map.end())
            return false;

        map<string, unsigned long long>::const_iterator seenNSIt = seenConnIt->second.find(ns);
        if (seenNSIt == seenConnIt->second.end())
            return false;

        *sequence = seenNSIt->second;
        return true;
    }

    void setSequence(DBClientBase* conn, const string& ns, const unsigned long long& s) {
        stdx::lock_guard<stdx::mutex> lk(_mutex);
        _map[conn->getConnectionId()][ns] = s;
    }

    void reset(DBClientBase* conn) {
        stdx::lock_guard<stdx::mutex> lk(_mutex);
        _map.erase(conn->getConnectionId());
    }

private:
    // protects _map
    stdx::mutex _mutex;

    // a map from a connection into ChunkManager's sequence number for each namespace
    typedef map<unsigned long long, map<string, unsigned long long>> SequenceMap;
    SequenceMap _map;

} connectionShardStatus;

/**
 * Sends the setShardVersion command on the specified connection.
 */
bool setShardVersion(OperationContext* opCtx,
                     DBClientBase* conn,
                     const string& ns,
                     const ConnectionString& configServer,
                     ChunkVersion version,
                     ChunkManager* manager,
                     bool authoritative,
                     BSONObj& result) {
    // This code should never run under a cross-shard transaction
    invariant(!TransactionRouter::get(opCtx));

    ShardId shardId;
    ConnectionString shardCS;
    {
        const auto shard = Grid::get(opCtx)->shardRegistry()->getShardForHostNoReload(
            uassertStatusOK(HostAndPort::parse(conn->getServerAddress())));
        uassert(ErrorCodes::ShardNotFound,
                str::stream() << conn->getServerAddress() << " is not recognized as a shard",
                shard);

        shardId = shard->getId();
        shardCS = shard->getConnString();
    }

    BSONObj cmd;

    if (ns.empty()) {
        SetShardVersionRequest ssv =
            SetShardVersionRequest::makeForInit(configServer, shardId, shardCS);
        cmd = ssv.toBSON();
    } else {
        SetShardVersionRequest ssv = SetShardVersionRequest::makeForVersioning(
            configServer, shardId, shardCS, NamespaceString(ns), version, authoritative);
        cmd = ssv.toBSON();
    }

    LOG(1) << "    setShardVersion  " << shardId << " " << conn->getServerAddress() << "  " << ns
           << "  " << cmd
           << (manager ? string(str::stream() << " " << manager->getSequenceNumber()) : "");

    return conn->runCommand("admin", cmd, result, 0);
}

/**
 * Checks whether the specified connection supports versioning.
 */
DBClientBase* getVersionable(DBClientBase* conn) {
    switch (conn->type()) {
        case ConnectionString::LOCAL:
        case ConnectionString::INVALID:
        case ConnectionString::CUSTOM:
            MONGO_UNREACHABLE;

        case ConnectionString::MASTER:
            return conn;
        case ConnectionString::SET:
            DBClientReplicaSet* set = (DBClientReplicaSet*)conn;
            return &(set->masterConn());
    }

    MONGO_UNREACHABLE;
}

/**
 * Special internal logic to run reduced version handshake for empty namespace operations to
 * shards.
 *
 * Eventually this should go completely away, but for now many commands rely on unversioned but
 * mongos-specific behavior on mongod (auditing and replication information in commands)
 */
bool initShardVersionEmptyNS(OperationContext* opCtx, DBClientBase* conn_in) {
    try {
        // May throw if replica set primary is down
        DBClientBase* const conn = getVersionable(conn_in);
        dassert(conn);  // errors thrown above

        // Check to see if we've already initialized this connection. This avoids sending
        // setShardVersion multiple times.
        if (connectionShardStatus.hasAnySequenceSet(conn)) {
            return false;
        }

        BSONObj result;
        const bool ok =
            setShardVersion(opCtx,
                            conn,
                            "",
                            Grid::get(opCtx)->shardRegistry()->getConfigServerConnectionString(),
                            ChunkVersion(),
                            nullptr,
                            true,
                            result);

        LOG(3) << "initial sharding result : " << result;

        connectionShardStatus.setSequence(conn, "", 0);
        return ok;
    } catch (const DBException&) {
        // NOTE: Replica sets may fail to initShardVersion because future calls relying on
        // correct versioning must later call checkShardVersion on the primary.
        // Secondary queries and commands may not call checkShardVersion, but secondary ops
        // aren't versioned at all.
        if (conn_in->type() != ConnectionString::SET) {
            throw;
        }

        // NOTE: Only old-style cluster operations will talk via DBClientReplicaSets - using
        // checkShardVersion is required (which includes initShardVersion information) if these
        // connections are used.

        static Occasionally sampler;
        if (sampler.tick()) {
            warning() << "failed to initialize new replica set connection version, "
                      << "will initialize on first use";
        }

        return false;
    }
}

/**
 * Updates the remote cached version on the remote shard host (primary, in the case of replica
 * sets) if needed with a fully-qualified shard version for the given namespace:
 *   config server(s) + shard name + shard version
 *
 * If no remote cached version has ever been set, an initial shard version is sent.
 *
 * If the namespace is empty and no version has ever been sent, the config server + shard name
 * is sent to the remote shard host to initialize the connection as coming from mongos.
 * NOTE: This initialization is *best-effort only*.  Operations which wish to correctly version
 * must send the namespace.
 *
 * Config servers are special and are not (unless otherwise a shard) kept up to date with this
 * protocol.  This is safe so long as config servers only contain unversioned collections.
 *
 * It is an error to call checkShardVersion with an unversionable connection (isVersionableCB).
 *
 * @return true if we contacted the remote host
 */
bool checkShardVersion(OperationContext* opCtx,
                       DBClientBase* conn_in,
                       const string& ns,
                       shared_ptr<ChunkManager> refManager,
                       bool authoritative,
                       int tryNumber) {
    // Empty namespaces are special - we require initialization but not versioning
    if (ns.size() == 0) {
        return initShardVersionEmptyNS(opCtx, conn_in);
    }

    DBClientBase* const conn = getVersionable(conn_in);
    verify(conn);  // errors thrown above

    const NamespaceString nss(ns);

    auto const catalogCache = Grid::get(opCtx)->catalogCache();

    if (authoritative) {
        Grid::get(opCtx)->catalogCache()->invalidateShardedCollection(nss);
    }

    auto routingInfoStatus = catalogCache->getCollectionRoutingInfo(opCtx, nss);
    if (!routingInfoStatus.isOK()) {
        return false;
    }

    auto& routingInfo = routingInfoStatus.getValue();

    const auto manager = routingInfo.cm();
    const auto primary = routingInfo.db().primary();

    unsigned long long officialSequenceNumber = 0;

    if (manager) {
        officialSequenceNumber = manager->getSequenceNumber();
    } else if (primary && primary->isConfig()) {
        // Do not send setShardVersion to collections on the config servers - this causes problems
        // when config servers are also shards and get SSV with conflicting names.
        return false;
    }

    const auto shardRegistry = Grid::get(opCtx)->shardRegistry();

    const auto shard = shardRegistry->getShardForHostNoReload(
        uassertStatusOK(HostAndPort::parse(conn->getServerAddress())));
    uassert(ErrorCodes::ShardNotFound,
            str::stream() << conn->getServerAddress() << " is not recognized as a shard",
            shard);

    // Check this manager against the reference manager
    if (manager) {
        if (refManager && !refManager->compatibleWith(*manager, shard->getId())) {
            const ChunkVersion refVersion(refManager->getVersion(shard->getId()));
            const ChunkVersion currentVersion(manager->getVersion(shard->getId()));

            string msg(str::stream() << "manager (" << currentVersion.toString() << " : "
                                     << manager->getSequenceNumber()
                                     << ") "
                                     << "not compatible with reference manager ("
                                     << refVersion.toString()
                                     << " : "
                                     << refManager->getSequenceNumber()
                                     << ") "
                                     << "on shard "
                                     << shard->getId()
                                     << " ("
                                     << shard->getConnString().toString()
                                     << ")");

            uasserted(StaleConfigInfo(nss, refVersion, currentVersion), msg);
        }
    } else if (refManager) {
        string msg(str::stream() << "not sharded (" << (!manager ? string("<none>") : str::stream()
                                                                << manager->getSequenceNumber())
                                 << ") but has reference manager ("
                                 << refManager->getSequenceNumber()
                                 << ") "
                                 << "on conn "
                                 << conn->getServerAddress()
                                 << " ("
                                 << conn_in->getServerAddress()
                                 << ")");

        uasserted(
            StaleConfigInfo(nss, refManager->getVersion(shard->getId()), ChunkVersion::UNSHARDED()),
            msg);
    }

    // Has the ChunkManager been reloaded since the last time we updated the shard version over
    // this connection?  If we've never updated the shard version, do so now.
    unsigned long long sequenceNumber = 0;
    if (connectionShardStatus.getSequence(conn, ns, &sequenceNumber)) {
        if (sequenceNumber == officialSequenceNumber) {
            return false;
        }
    }

    ChunkVersion version = ChunkVersion(0, 0, OID());
    if (manager) {
        version = manager->getVersion(shard->getId());
    }

    LOG(1) << "setting shard version of " << version << " for " << ns << " on shard "
           << shard->toString();

    LOG(3) << "last version sent with chunk manager iteration " << sequenceNumber
           << ", current chunk manager iteration is " << officialSequenceNumber;

    BSONObj result;
    if (setShardVersion(opCtx,
                        conn,
                        ns,
                        shardRegistry->getConfigServerConnectionString(),
                        version,
                        manager.get(),
                        authoritative,
                        result)) {
        LOG(1) << "      setShardVersion success: " << result;
        connectionShardStatus.setSequence(conn, ns, officialSequenceNumber);
        return true;
    }

    // If the shard rejected the setShardVersion, return the error to the user.
    int errCode = result["code"].numberInt();
    uassert(errCode, result["errmsg"].String(), errCode != ErrorCodes::NoShardingEnabled);

    LOG(1) << "       setShardVersion failed!\n" << result;

    if (result["need_authoritative"].trueValue())
        massert(10428, "need_authoritative set but in authoritative mode already", !authoritative);

    if (!authoritative) {
        // use the original connection and get a fresh versionable connection
        // since conn can be invalidated (or worse, freed) after the failure
        checkShardVersion(opCtx, conn_in, ns, refManager, 1, tryNumber + 1);
        return true;
    }

    Grid::get(opCtx)->catalogCache()->onStaleShardVersion(std::move(routingInfo));

    const int maxNumTries = 7;
    if (tryNumber < maxNumTries) {
        LOG(tryNumber < (maxNumTries / 2) ? 1 : 0)
            << "going to retry checkShardVersion shard: " << shard->toString() << " " << result;
        sleepmillis(10 * tryNumber);
        // use the original connection and get a fresh versionable connection
        // since conn can be invalidated (or worse, freed) after the failure
        checkShardVersion(opCtx, conn_in, ns, refManager, true, tryNumber + 1);
        return true;
    }

    string errmsg = str::stream() << "setShardVersion failed shard: " << shard->toString() << " "
                                  << result;
    log() << "     " << errmsg;
    massert(10429, errmsg, 0);
    return true;
}

}  // namespace

// Global version manager
VersionManager versionManager;

void VersionManager::resetShardVersionCB(DBClientBase* conn) {
    connectionShardStatus.reset(conn);
}

bool VersionManager::isVersionableCB(DBClientBase* conn) {
    // We do not version shard connections when issued from mongod
    if (!isMongos()) {
        return false;
    }

    return conn->type() == ConnectionString::MASTER || conn->type() == ConnectionString::SET;
}

bool VersionManager::checkShardVersionCB(OperationContext* opCtx,
                                         DBClientBase* conn_in,
                                         const string& ns,
                                         bool authoritative,
                                         int tryNumber) {
    return checkShardVersion(opCtx, conn_in, ns, nullptr, authoritative, tryNumber);
}

bool VersionManager::checkShardVersionCB(OperationContext* opCtx,
                                         ShardConnection* conn_in,
                                         bool authoritative,
                                         int tryNumber) {
    return checkShardVersion(
        opCtx, conn_in->get(), conn_in->getNS(), conn_in->getManager(), authoritative, tryNumber);
}

}  // namespace mongo