Merge pull request #154 from sandeepsnairms/main

Rebranding + Resolving Conflicts

Author: sandeepsnairms

MongoMigrationWebApp/Program.cs

@@ -118,8 +118,8 @@ CRITICAL ERROR - {source}
     return client;
 });
 
-
-builder.Services.AddRazorPages();
+    builder.Configuration.AddEnvironmentVariables();
+    builder.Services.AddRazorPages();
 builder.Services.AddServerSideBlazor();
 builder.Services.AddSingleton(builder.Configuration);
 builder.Services.AddSingleton<JobManager>();

MongoMigrationWebApp/Service/JobManager.cs

@@ -1,21 +1,67 @@
-using System;
-using System.Collections.Generic;
+using Microsoft.Extensions.Configuration;
 using OnlineMongoMigrationProcessor;
+using OnlineMongoMigrationProcessor.Helpers;
 using OnlineMongoMigrationProcessor.Models;
 using OnlineMongoMigrationProcessor.Workers;
-using OnlineMongoMigrationProcessor.Helpers;
+using System;
+using System.Collections.Generic;
+
 
 namespace MongoMigrationWebApp.Service
 {
 
     public class JobManager
     {
-    private JobList? _jobList;
+        private JobList? _jobList;
         private MigrationWorker? MigrationWorker { get; set; }
 
         private DateTime _lastJobHeartBeat = DateTime.MinValue;
         private string _lastJobID = string.Empty;
 
+        public JobManager(IConfiguration configuration)
+        {
+            System.IO.File.AppendAllText($"{Helper.GetWorkingFolder()}logabc.txt", $"Resuming migration job after application restart...");
+            var migrationJobs = GetMigrations(out string errorMessage);
+
+            System.IO.File.AppendAllText($"{Helper.GetWorkingFolder()}logabc.txt", $"Step 0");
+
+            if (migrationJobs != null && migrationJobs.Count == 1)
+            {
+
+                System.IO.File.AppendAllText($"{Helper.GetWorkingFolder()}logabc.txt", $"Step1 : {migrationJobs[0].IsStarted}-{migrationJobs[0].IsCompleted} -{string.IsNullOrEmpty(migrationJobs[0].SourceConnectionString)} - {string.IsNullOrEmpty(migrationJobs[0].TargetConnectionString)} ");
+                if (migrationJobs[0].IsStarted && !migrationJobs[0].IsCompleted && string.IsNullOrEmpty(migrationJobs[0].SourceConnectionString) && string.IsNullOrEmpty(migrationJobs[0].TargetConnectionString))
+                {
+                    try
+                    {
+                        System.IO.File.AppendAllText($"{Helper.GetWorkingFolder()}logabc.txt", $"Step2 : before reading config"); 
+
+
+                        var sourceConnectionString = configuration.GetConnectionString("SourceConnectionString");
+                        var targetConnectionString = configuration.GetConnectionString("TargetConnectionString");
+
+                        if(sourceConnectionString != null && targetConnectionString != null)
+                            System.IO.File.AppendAllText($"{Helper.GetWorkingFolder()}logabc.txt", $"Step 3 :Cluster found" + sourceConnectionString.Contains("cluster"));
+
+                        var tmpSrcEndpoint = Helper.ExtractHost(sourceConnectionString);
+                        var tmpTgtEndpoint = Helper.ExtractHost(targetConnectionString);
+                        if (migrationJobs[0].SourceEndpoint == tmpSrcEndpoint && migrationJobs[0].TargetEndpoint == tmpTgtEndpoint)
+                        {
+
+                            migrationJobs[0].SourceConnectionString = sourceConnectionString;
+                            migrationJobs[0].TargetConnectionString = targetConnectionString;
+                            //ViewMigration(migrationJobs[0].Id);
+                            StartMigrationAsync(migrationJobs[0], sourceConnectionString, targetConnectionString, migrationJobs[0].NameSpaces, migrationJobs[0].JobType, Helper.IsOnline(migrationJobs[0]));
+                        }
+                    }
+                    catch(Exception ex)
+                    {
+                        System.IO.File.AppendAllText($"{Helper.GetWorkingFolder()}logabc.txt", $"Exception : {ex}"); //DO NOTHING
+
+                    }
+                }
+            }
+        }
+
         #region _configuration Management
 
         public bool UpdateConfig(OnlineMongoMigrationProcessor.MigrationSettings updated_config, out string errorMessage)

MongoMigrationWebApp/Shared/MainLayout.razor

@@ -5,7 +5,7 @@
 <div class="page">
     <main>
         <div class="top-row px-4 d-flex justify-content-between align-items-center">
-            <h2 class="text-left">Migrate to Azure Cosmos DB for MongoDB (vCore based)</h2>
+            <h2 class="text-left">Migrate to Azure DocumentDB</h2>
             <div class="ms-2 small">v0.9.2.5.2</div>
         </div>
 

OnlineMongoMigrationProcessor/Helpers/AccumulatedChangesTracker.cs

@@ -14,23 +14,39 @@ public class AccumulatedChangesTracker
         public List<ChangeStreamDocument<BsonDocument>> DocsToBeUpdated { get; private set; } = new();
         public List<ChangeStreamDocument<BsonDocument>> DocsToBeDeleted { get; private set; } = new();
 
-        
+
         // Track the latest resume token for checkpoint on success
+        public string CollectionKey { get; private set; } = string.Empty;
         public string LatestResumeToken { get; private set; } = string.Empty;
         public DateTime LatestTimestamp { get; private set; } = DateTime.MinValue;
         public ChangeStreamOperationType LatestOperationType { get; private set; }
         public string LatestDocumentKey { get; private set; } = string.Empty;
 
+        private string _collectionKey = string.Empty;
+        public AccumulatedChangesTracker(string collectionKey)
+        {
+            _collectionKey = collectionKey;
+            CollectionKey = _collectionKey;
+        }
+
+
         public void AddInsert(ChangeStreamDocument<BsonDocument> change)
         {
+            var changeCollectionKey = $"{change.DatabaseNamespace?.DatabaseName}.{change.CollectionNamespace?.CollectionName}";
+            if (changeCollectionKey != _collectionKey)
+                return;
+
             var id = change.DocumentKey.ToJson();
 
+            if(string.IsNullOrEmpty(id))
+                return;
+
             // Remove from other lists
-            DocsToBeUpdated.RemoveAll(c => c.DocumentKey.ToJson() == id);
-            DocsToBeDeleted.RemoveAll(c => c.DocumentKey.ToJson() == id);
+            DocsToBeUpdated.RemoveAll(c => c.DocumentKey != null && c.DocumentKey.ToJson() == id);
+            DocsToBeDeleted.RemoveAll(c => c.DocumentKey != null && c.DocumentKey.ToJson() == id);
 
             // Replace if already exists
-            DocsToBeInserted.RemoveAll(c => c.DocumentKey.ToJson() == id);
+            DocsToBeInserted.RemoveAll(c => c.DocumentKey != null && c.DocumentKey.ToJson() == id);
             DocsToBeInserted.Add(change);
 
             // Track earliest and latest change metadata for checkpoint updates
@@ -39,13 +55,20 @@ public void AddInsert(ChangeStreamDocument<BsonDocument> change)
 
         public void AddUpdate(ChangeStreamDocument<BsonDocument> change)
         {
+            var changeCollectionKey = $"{change.DatabaseNamespace?.DatabaseName}.{change.CollectionNamespace?.CollectionName}";
+            if (changeCollectionKey != _collectionKey)
+                return;
+
             var id = change.DocumentKey.ToJson();
 
+            if (string.IsNullOrEmpty(id))
+                return;
+
             // Remove from delete list
-            DocsToBeDeleted.RemoveAll(c => c.DocumentKey.ToJson() == id);
+            DocsToBeDeleted.RemoveAll(c => c.DocumentKey != null && c.DocumentKey != null && c.DocumentKey.ToJson() == id);
 
             // Replace in update list
-            DocsToBeUpdated.RemoveAll(c => c.DocumentKey.ToJson() == id);
+            DocsToBeUpdated.RemoveAll(c => c.DocumentKey != null && c.DocumentKey.ToJson() == id);
             DocsToBeUpdated.Add(change);
 
             // Don't remove from insert — updates after insert are valid
@@ -56,14 +79,21 @@ public void AddUpdate(ChangeStreamDocument<BsonDocument> change)
 
         public void AddDelete(ChangeStreamDocument<BsonDocument> change)
         {
+            var changeCollectionKey = $"{change.DatabaseNamespace?.DatabaseName}.{change.CollectionNamespace?.CollectionName}";
+            if (changeCollectionKey != _collectionKey)
+                return;
+
             var id = change.DocumentKey.ToJson();
 
+            if (string.IsNullOrEmpty(id))
+                return;
+
             // Remove from insert and update
-            DocsToBeInserted.RemoveAll(c => c.DocumentKey.ToJson() == id);
-            DocsToBeUpdated.RemoveAll(c => c.DocumentKey.ToJson() == id);
+            DocsToBeInserted.RemoveAll(c => c.DocumentKey != null && c.DocumentKey != null && c.DocumentKey.ToJson() == id);
+            DocsToBeUpdated.RemoveAll(c => c.DocumentKey != null && c.DocumentKey != null && c.DocumentKey.ToJson() == id);
 
             // Replace in delete list
-            DocsToBeDeleted.RemoveAll(c => c.DocumentKey.ToJson() == id);
+            DocsToBeDeleted.RemoveAll(c => c.DocumentKey != null && c.DocumentKey != null && c.DocumentKey.ToJson() == id);
             DocsToBeDeleted.Add(change);
 
             // Track earliest and latest change metadata for checkpoint updates

OnlineMongoMigrationProcessor/Helpers/MongoHelper.cs

@@ -632,6 +632,7 @@ public async static Task SetChangeStreamResumeTokenAsync(Log log, MongoClient cl
         int retryCount = 0;
         bool isSucessful = false;
         bool resetCS = unit.ResetChangeStream;
+        bool useServerLevel=false;
 
         while (!isSucessful && retryCount < 10)
         {
@@ -642,7 +643,7 @@ public async static Task SetChangeStreamResumeTokenAsync(Log log, MongoClient cl
                 var collection = database.GetCollection<BsonDocument>(unit.CollectionName);
 
                 // Determine if we should use server-level or collection-level processing
-                bool useServerLevel = job.ChangeStreamLevel == ChangeStreamLevel.Server;
+                useServerLevel = job.ChangeStreamLevel == ChangeStreamLevel.Server;
 
                 // Initialize with safe defaults; will be overridden below
                 var options = new ChangeStreamOptions { BatchSize = 500, FullDocument = ChangeStreamFullDocumentOption.UpdateLookup };
@@ -696,24 +697,55 @@ public async static Task SetChangeStreamResumeTokenAsync(Log log, MongoClient cl
                     };
                 }
 
-                //new way to get resume token
-                //On MongoDB 4.0+, the WatchChangeStreamAsync method opens a change stream and waits for changes.
-                //On MongoDB 3.6, TailOplogAsync opens a tailable cursor on the oplog, filtering on namespace and timestamp to detect new operations.
-                if (!string.IsNullOrEmpty(job.SourceServerVersion) && job.SourceServerVersion.StartsWith("3"))
-                {
-                    if (!await TailOplogAsync(client, unit.DatabaseName, unit.CollectionName, unit, cts) && !resetCS)
+                    //new way to get resume token
+                    //On MongoDB 4.0+, the WatchChangeStreamAsync method opens a change stream and waits for changes.
+                    //On MongoDB 3.6, TailOplogAsync opens a tailable cursor on the oplog, filtering on namespace and timestamp to detect new operations.
+                    if (!string.IsNullOrEmpty(job.SourceServerVersion) && job.SourceServerVersion.StartsWith("3"))
                     {
-                        //if failed to tail oplog, fallback to watching change stream infinelty. Should be called async only
-                        await WatchChangeStreamUntilChangeAsync(log, client, jobList, job, unit, collection, options, resetCS, -1, cts, useServerLevel);
+                        if (!await TailOplogAsync(client, unit.DatabaseName, unit.CollectionName, unit, cts) && !resetCS)
+                        {
+                            //if failed to tail oplog, fallback to watching change stream infinelty. Should be called async only
+                            await WatchChangeStreamUntilChangeAsync(log, client, jobList, job, unit, collection, options, resetCS, -1, cts, useServerLevel);//don't await
+                        }
+                    }
+                    else
+                    {
+                        //await WatchChangeStreamUntilChangeAsync(log, client, jobList, job, unit, collection, options, resetCS, seconds, cts, useServerLevel);
+                        //end of new way to get resume token
+
+                        var settings = client.Settings.Clone();
+                        var localClient = new MongoClient(settings);
+
+                        var result = await MongoSafeTaskExecutor.ExecuteAsync(
+                        operation: async (ct) =>
+                        {
+                            // forward the cancellation token to your watch function
+                            return await WatchChangeStreamUntilChangeAsync(
+                                log,
+                                localClient,
+                                jobList,
+                                job,
+                                unit,
+                                collection,
+                                options,
+                                resetCS,
+                                seconds,
+                                ct,
+                                useServerLevel
+                            );
+                        },
+                        timeoutSeconds: seconds +10,   // choose your timeout
+                        operationName: $"WatchChangeStreamUntilChangeAsync({collection.CollectionNamespace.CollectionName})",
+                        logAction: msg => log.WriteLine(msg,LogType.Debug),
+                        externalToken: cts,
+                        clientToKill: localClient    // <-- VERY IMPORTANT: kill this client if watch hangs
+                        );
+
                     }
-                }
-                else
-                    await WatchChangeStreamUntilChangeAsync(log, client, jobList, job, unit, collection, options, resetCS, seconds, cts, useServerLevel);
-                //end of new way to get resume token
 
                 isSucessful = true;
             }
-            catch (OperationCanceledException)
+            catch (Exception ex) when (ex is OperationCanceledException || ex is TimeoutException)
             {
                 // Distinguish between timeout (no activity) vs manual cancellation
                 if (resetCS && unit.ResetChangeStream)
@@ -764,12 +796,17 @@ public async static Task SetChangeStreamResumeTokenAsync(Log log, MongoClient cl
             finally
             {
                 jobList.Save();
+                if (useServerLevel)
+                   log.WriteLine($"Exiting Server-level SetChangeStreamResumeTokenAsync for job {job.Id}", LogType.Info);
+                else
+                    log.WriteLine($"Exiting Collection-level SetChangeStreamResumeToken for {unit.DatabaseName}.{unit.CollectionName}", LogType.Debug);
             }
+
         }
         return;
     }
 
-    private static async Task WatchChangeStreamUntilChangeAsync(Log log, MongoClient client, JobList jobList, MigrationJob job, MigrationUnit unit, IMongoCollection<BsonDocument> collection, ChangeStreamOptions options, bool resetCS, int seconds, CancellationToken manualCts, bool useServerLevel = false)
+    private static async Task<bool> WatchChangeStreamUntilChangeAsync(Log log, MongoClient client, JobList jobList, MigrationJob job, MigrationUnit unit, IMongoCollection<BsonDocument> collection, ChangeStreamOptions options, bool resetCS, int seconds, CancellationToken manualCts, bool useServerLevel = false)
     {
         var pipeline = new BsonDocument[] { };
         if (job.JobType == JobType.RUOptimizedCopy)
@@ -810,13 +847,13 @@ private static async Task WatchChangeStreamUntilChangeAsync(Log log, MongoClient
         if (useServerLevel)
         {
             // Server-level change stream
-            log.WriteLine($"Setting up server-level change stream resume token for job {job.Id}");
+            log.WriteLine($"Setting up server-level change stream resume token for job {job.Id}", LogType.Debug);
             cursor = await client.WatchAsync<ChangeStreamDocument<BsonDocument>>(pipeline, options, linkedCts.Token);
         }
         else
         {
             // Collection-level change stream
-            log.WriteLine($"Setting up collection-level change stream resume token for {unit.DatabaseName}.{unit.CollectionName}");
+            log.WriteLine($"Setting up collection-level change stream resume token for {unit.DatabaseName}.{unit.CollectionName}", LogType.Debug);
             cursor = await collection.WatchAsync<ChangeStreamDocument<BsonDocument>>(pipeline, options, linkedCts.Token);
         }
 
@@ -839,9 +876,9 @@ private static async Task WatchChangeStreamUntilChangeAsync(Log log, MongoClient
                             unit.OriginalResumeToken = resumeTokenJson;
 
                         }
-                        return;
+                        return true;
                     }
-                    return;
+                    return true;
                 }                   
 
                 // Iterate until cancellation or first change detected
@@ -859,7 +896,7 @@ private static async Task WatchChangeStreamUntilChangeAsync(Log log, MongoClient
                         //if bulk load is complete, no point in continuing to watch
                         // Use the local resetCS variable captured at method start, not unit.ResetChangeStream
                         if ((unit.RestoreComplete || job.IsSimulatedRun) && unit.DumpComplete && !resetCS)
-                            return;
+                            return true;
 
 
                         // Handle server-level vs collection-level resume token storage
@@ -881,7 +918,7 @@ private static async Task WatchChangeStreamUntilChangeAsync(Log log, MongoClient
 
                                 log.WriteLine($"Server-level resume token set for job {job.Id} with collection key {job.ResumeCollectionKey}");
                                 // Exit immediately after first change detected
-                                return;
+                                return true;
                             }
                         }
                         else
@@ -892,16 +929,19 @@ private static async Task WatchChangeStreamUntilChangeAsync(Log log, MongoClient
                             log.WriteLine($"Collection-level resume token set for {unit.DatabaseName}.{unit.CollectionName}");
 
                             // Exit immediately after first change detected
-                            return;
+                            return true;
                         }
 
                     }
                 }
-                   
+
+               return false;
             }
             catch (OperationCanceledException)
             {
                 // Cancellation requested - exit quietly
+                log.WriteLine($"Collection-level resume token request for {unit.DatabaseName}.{unit.CollectionName} was canceled.", LogType.Debug);
+                return false;
             }
         }
     }