ydb/core/cms/console/console__cleanup_subscriptions.cpp


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

#include "config_index.h"
#include "console_impl.h"
#include "console_configs_manager.h"
#include "console_tenants_manager.h"

#include <ydb/core/base/path.h>
#include <ydb/core/cms/console/validators/registry.h>

namespace NKikimr::NConsole {

class TConfigsManager::TTxCleanupSubscriptions : public TTransactionBase<TConfigsManager> {
public:
    TTxCleanupSubscriptions(TEvInterconnect::TEvNodesInfo::TPtr &ev,
                            TConfigsManager *self)
        : TBase(self)
        , Nodes(ev)
    {
    }

    bool Execute(TTransactionContext &txc, const TActorContext &ctx) override
    {
        LOG_DEBUG_S(ctx, NKikimrServices::CMS_CONFIGS,
                    "TConsole::TTxCleanupSubscriptions");

        THashSet<ui32> nodes;
        for (auto &node : Nodes->Get()->Nodes)
            nodes.insert(node.NodeId);

        for (auto pr : Self->SubscriptionIndex.GetSubscriptions()) {
            auto nodeId = pr.second->Subscriber.ServiceId.NodeId();
            if (nodeId && !nodes.contains(nodeId)) {
                Self->PendingSubscriptionModifications.RemovedSubscriptions.insert(pr.first);

                LOG_DEBUG_S(ctx, NKikimrServices::CMS_CONFIGS,
                            "Subscription " << pr.first << " has subscriber from unknown node "
                            << nodeId << " and will be removed");
            }
        }

        // Update database.
        Self->DbApplyPendingSubscriptionModifications(txc, ctx);

        return true;
    }

    void Complete(const TActorContext &executorCtx) override
    {
        auto ctx = executorCtx.MakeFor(Self->SelfId());
        LOG_DEBUG(ctx, NKikimrServices::CMS_CONFIGS,
                  "TConsole::TTxCleanupSubscriptions Complete");

        if (!Self->PendingSubscriptionModifications.IsEmpty())
            Self->ApplyPendingSubscriptionModifications(ctx);
        Self->ScheduleSubscriptionsCleanup(ctx);

        Self->TxProcessor->TxCompleted(this, ctx);
    }

private:
    TEvInterconnect::TEvNodesInfo::TPtr Nodes;
};

ITransaction *TConfigsManager::CreateTxCleanupSubscriptions(TEvInterconnect::TEvNodesInfo::TPtr &ev)
{
    return new TTxCleanupSubscriptions(ev, this);
}

} // namespace NKikimr::NConsole