Async replication: configurable default retention period (#8270)

Author: CyberROFL

ydb/core/base/appdata.cpp

@@ -20,6 +20,7 @@
 #include <ydb/core/protos/key.pb.h>
 #include <ydb/core/protos/memory_controller_config.pb.h>
 #include <ydb/core/protos/pqconfig.pb.h>
+#include <ydb/core/protos/replication.pb.h>
 #include <ydb/core/protos/stream.pb.h>
 #include <ydb/core/protos/netclassifier.pb.h>
 #include <ydb/core/protos/datashard_config.pb.h>
@@ -63,6 +64,7 @@ struct TAppData::TImpl {
     NKikimrSharedCache::TSharedCacheConfig SharedCacheConfig;
     NKikimrConfig::TMetadataCacheConfig MetadataCacheConfig;
     NKikimrConfig::TMemoryControllerConfig MemoryControllerConfig;
+    NKikimrReplication::TReplicationDefaults ReplicationConfig;
 };
 
 TAppData::TAppData(
@@ -116,6 +118,7 @@ TAppData::TAppData(
     , SharedCacheConfig(Impl->SharedCacheConfig)
     , MetadataCacheConfig(Impl->MetadataCacheConfig)
     , MemoryControllerConfig(Impl->MemoryControllerConfig)
+    , ReplicationConfig(Impl->ReplicationConfig)
     , KikimrShouldContinue(kikimrShouldContinue)
 {}
 

ydb/core/base/appdata_fwd.h

@@ -66,6 +66,10 @@ namespace NKikimrConfig {
     class TMemoryControllerConfig;
 }
 
+namespace NKikimrReplication {
+    class TReplicationDefaults;
+}
+
 namespace NKikimrNetClassifier {
     class TNetClassifierDistributableConfig;
     class TNetClassifierConfig;
@@ -215,6 +219,7 @@ struct TAppData {
     NKikimrSharedCache::TSharedCacheConfig& SharedCacheConfig;
     NKikimrConfig::TMetadataCacheConfig& MetadataCacheConfig;
     NKikimrConfig::TMemoryControllerConfig& MemoryControllerConfig;
+    NKikimrReplication::TReplicationDefaults& ReplicationConfig;
     bool EnforceUserTokenRequirement = false;
     bool EnforceUserTokenCheckRequirement = false; // check token if it was specified
     bool AllowHugeKeyValueDeletes = true; // delete when all clients limit deletes per request

ydb/core/driver_lib/run/run.cpp

@@ -1131,6 +1131,10 @@ void TKikimrRunner::InitializeAppData(const TKikimrRunConfig& runConfig)
         AppData->MemoryControllerConfig.CopyFrom(runConfig.AppConfig.GetMemoryControllerConfig());
     }
 
+    if (runConfig.AppConfig.HasReplicationConfig()) {
+        AppData->ReplicationConfig = runConfig.AppConfig.GetReplicationConfig();
+    }
+
     // setup resource profiles
     AppData->ResourceProfiles = new TResourceProfiles;
     if (runConfig.AppConfig.GetBootstrapConfig().ResourceProfilesSize())

ydb/core/protos/config.proto

@@ -25,6 +25,7 @@ import "ydb/core/protos/memory_controller_config.proto";
 import "ydb/core/protos/netclassifier.proto";
 import "ydb/core/protos/node_broker.proto";
 import "ydb/core/protos/pqconfig.proto";
+import "ydb/core/protos/replication.proto";
 import "ydb/core/protos/resource_broker.proto";
 import "ydb/core/protos/shared_cache.proto";
 import "ydb/core/protos/stream.proto";
@@ -1946,6 +1947,7 @@ message TAppConfig {
     optional TMetadataCacheConfig MetadataCacheConfig = 80;
     optional TMemoryControllerConfig MemoryControllerConfig = 81;
 	optional TGroupedMemoryLimiterConfig GroupedMemoryLimiterConfig = 82;
+    optional NKikimrReplication.TReplicationDefaults ReplicationConfig = 83;
 
     repeated TNamedConfig NamedConfigs = 100;
     optional string ClusterYamlConfig = 101;

ydb/core/protos/console_config.proto

@@ -140,6 +140,7 @@ message TConfigItem {
         BackgroundCleaningConfigItem = 77;
         MetadataCacheConfigItem = 80;
         MemoryControllerConfigItem = 81;
+        ReplicationConfigItem = 83;
 
         NamedConfigsItem = 100;
         ClusterYamlConfigItem = 101;

ydb/core/protos/replication.proto

@@ -5,6 +5,10 @@ import "ydb/public/api/protos/ydb_issue_message.proto";
 package NKikimrReplication;
 option java_package = "ru.yandex.kikimr.proto";
 
+message TReplicationDefaults {
+    optional int32 RetentionPeriodSeconds = 1 [default = 86400]; // 1d
+}
+
 message TStaticCredentials {
     optional string User = 1;
     optional string Password = 2 [(Ydb.sensitive) = true];

ydb/core/tx/replication/controller/stream_creator.cpp

@@ -4,7 +4,9 @@
 #include "target_with_stream.h"
 #include "util.h"
 
+#include <ydb/core/base/appdata.h>
 #include <ydb/core/base/path.h>
+#include <ydb/core/protos/replication.pb.h>
 #include <ydb/core/tx/replication/ydb_proxy/ydb_proxy.h>
 #include <ydb/library/actors/core/actor_bootstrapped.h>
 #include <ydb/library/actors/core/hfunc.h>
@@ -16,9 +18,12 @@
 namespace NKikimr::NReplication::NController {
 
 class TStreamCreator: public TActorBootstrapped<TStreamCreator> {
-    static NYdb::NTable::TChangefeedDescription MakeChangefeed(const TString& name, const NJson::TJsonMap& attrs) {
+    static NYdb::NTable::TChangefeedDescription MakeChangefeed(
+            const TString& name, const TDuration& retentionPeriod, const NJson::TJsonMap& attrs)
+    {
         using namespace NYdb::NTable;
         return TChangefeedDescription(name, EChangefeedMode::Updates, EChangefeedFormat::Json)
+            .WithRetentionPeriod(retentionPeriod)
             .WithInitialScan()
             .AddAttribute("__async_replication", NJson::WriteJson(attrs, false));
     }
@@ -133,14 +138,15 @@ class TStreamCreator: public TActorBootstrapped<TStreamCreator> {
             TReplication::ETargetKind kind,
             const TString& srcPath,
             const TString& dstPath,
-            const TString& streamName)
+            const TString& streamName,
+            const TDuration& streamRetentionPeriod)
         : Parent(parent)
         , YdbProxy(proxy)
         , ReplicationId(rid)
         , TargetId(tid)
         , Kind(kind)
         , SrcPath(srcPath)
-        , Changefeed(MakeChangefeed(streamName, NJson::TJsonMap{
+        , Changefeed(MakeChangefeed(streamName, streamRetentionPeriod, NJson::TJsonMap{
             {"path", dstPath},
             {"id", ToString(rid)},
         }))
@@ -175,13 +181,15 @@ IActor* CreateStreamCreator(TReplication* replication, ui64 targetId, const TAct
     Y_ABORT_UNLESS(target);
     return CreateStreamCreator(ctx.SelfID, replication->GetYdbProxy(),
         replication->GetId(), target->GetId(), target->GetKind(),
-        target->GetSrcPath(), target->GetDstPath(), target->GetStreamName());
+        target->GetSrcPath(), target->GetDstPath(), target->GetStreamName(),
+        TDuration::Seconds(AppData()->ReplicationConfig.GetRetentionPeriodSeconds()));
 }
 
 IActor* CreateStreamCreator(const TActorId& parent, const TActorId& proxy, ui64 rid, ui64 tid,
-        TReplication::ETargetKind kind, const TString& srcPath, const TString& dstPath, const TString& streamName)
+        TReplication::ETargetKind kind, const TString& srcPath, const TString& dstPath,
+        const TString& streamName, const TDuration& streamRetentionPeriod)
 {
-    return new TStreamCreator(parent, proxy, rid, tid, kind, srcPath, dstPath, streamName);
+    return new TStreamCreator(parent, proxy, rid, tid, kind, srcPath, dstPath, streamName, streamRetentionPeriod);
 }
 
 }

ydb/core/tx/replication/controller/stream_creator.h

@@ -6,6 +6,7 @@ namespace NKikimr::NReplication::NController {
 
 IActor* CreateStreamCreator(TReplication* replication, ui64 targetId, const TActorContext& ctx);
 IActor* CreateStreamCreator(const TActorId& parent, const TActorId& proxy, ui64 rid, ui64 tid,
-    TReplication::ETargetKind kind, const TString& srcPath, const TString& dstPath, const TString& streamName);
+    TReplication::ETargetKind kind, const TString& srcPath, const TString& dstPath,
+    const TString& streamName, const TDuration& streamRetentionPeriod);
 
 }