ydb/yt/cpp/mapreduce/client/operation_helpers.cpp

#include "operation_helpers.h"

#include <yt/cpp/mapreduce/common/retry_lib.h>

#include <yt/cpp/mapreduce/http/context.h>
#include <yt/cpp/mapreduce/http/requests.h>
#include <yt/cpp/mapreduce/http/retry_request.h>

#include <yt/cpp/mapreduce/interface/config.h>
#include <yt/cpp/mapreduce/interface/raw_client.h>

#include <yt/cpp/mapreduce/interface/logging/yt_log.h>

#include <util/string/builder.h>

#include <util/system/mutex.h>
#include <util/system/rwlock.h>

namespace NYT::NDetail {

////////////////////////////////////////////////////////////////////////////////

ui64 RoundUpFileSize(ui64 size)
{
    constexpr ui64 roundUpTo = 4ull << 10;
    return (size + roundUpTo - 1) & ~(roundUpTo - 1);
}

bool UseLocalModeOptimization(
    const IRawClientPtr& rawClient,
    const TClientContext& context,
    const IClientRetryPolicyPtr& clientRetryPolicy)
{
    if (!context.Config->EnableLocalModeOptimization) {
        return false;
    }

    static THashMap<TString, bool> localModeMap;
    static TRWMutex mutex;

    {
        TReadGuard guard(mutex);
        auto it = localModeMap.find(context.ServerName);
        if (it != localModeMap.end()) {
            return it->second;
        }
    }

    bool isLocalMode = false;
    TString localModeAttr("//sys/@local_mode_fqdn");
    // We don't want to pollute logs with errors about failed request,
    // so we check if path exists before getting it.
    auto exists = RequestWithRetry<bool>(
        clientRetryPolicy->CreatePolicyForGenericRequest(),
        [&rawClient, &localModeAttr] (TMutationId /*mutationId*/) {
            return rawClient->Exists(
                TTransactionId(),
                localModeAttr,
                TExistsOptions().ReadFrom(EMasterReadKind::Cache));
        });
    if (exists)
    {
        auto fqdnNode = RequestWithRetry<TNode>(
            clientRetryPolicy->CreatePolicyForGenericRequest(),
            [&rawClient, &localModeAttr] (TMutationId /*mutationId*/) {
                return rawClient->TryGet(
                    TTransactionId(),
                    localModeAttr,
                    TGetOptions().ReadFrom(EMasterReadKind::Cache));
            });
        if (!fqdnNode.IsUndefined()) {
            auto fqdn = fqdnNode.AsString();
            isLocalMode = (fqdn == TProcessState::Get()->FqdnHostName);
            YT_LOG_DEBUG("Checking local mode; LocalModeFqdn: %v FqdnHostName: %v IsLocalMode: %v",
                fqdn,
                TProcessState::Get()->FqdnHostName,
                isLocalMode ? "true" : "false");
        }
    }

    {
        TWriteGuard guard(mutex);
        localModeMap[context.ServerName] = isLocalMode;
    }

    return isLocalMode;
}

TString GetOperationWebInterfaceUrl(TStringBuf serverName, TOperationId operationId)
{
    serverName.ChopSuffix(":80");
    serverName.ChopSuffix(".yt.yandex-team.ru");
    serverName.ChopSuffix(".yt.yandex.net");
    return ::TStringBuilder() << "https://yt.yandex-team.ru/" << serverName <<
        "/operations/" << GetGuidAsString(operationId);
}

////////////////////////////////////////////////////////////////////////////////

} // namespace NYT::NDetail