Bring back sync functionality (#1278)

## Changes
* Bring back sync functionality. Users can choose to manually start sync
from the UI and not rely on deploy before file runs. This avoids
breaking some existing user flows.
* The files are still run only after a deploy.
* There is possibility of race conditions between sync and deploy but
the possibility is very small. Also, syncing is blocked for prod and
staging targets.
 
## Tests
<!-- How is this tested? -->

Author: kartikgupta-db

packages/databricks-vscode/package.json

@@ -322,6 +322,27 @@
                 "title": "Force destroy bundle",
                 "category": "Databricks",
                 "enablement": "databricks.context.activated && databricks.context.bundle.isTargetSet && databricks.context.bundle.deploymentState == idle"
+            },
+            {
+                "command": "databricks.sync.start",
+                "title": "Start synchronization",
+                "category": "Databricks",
+                "enablement": "databricks.context.activated && databricks.context.loggedIn && databricks.context.bundle.isTargetSet && databricks.context.bundle.isDevTarget",
+                "icon": "$(sync)"
+            },
+            {
+                "command": "databricks.sync.startFull",
+                "title": "Start synchronization (full sync)",
+                "category": "Databricks",
+                "enablement": "databricks.context.activated && databricks.context.loggedIn && databricks.context.bundle.isTargetSet && databricks.context.bundle.isDevTarget",
+                "icon": "$(sync)"
+            },
+            {
+                "command": "databricks.sync.stop",
+                "title": "Stop synchronization",
+                "category": "Databricks",
+                "enablement": "databricks.context.activated && databricks.context.loggedIn && databricks.context.bundle.isTargetSet && databricks.context.bundle.isDevTarget",
+                "icon": "$(sync-ignored)"
             }
         ],
         "viewsContainers": {
@@ -538,6 +559,16 @@
                     "when": "view == configurationView && viewItem =~ /^databricks.environment.checkEnvironmentDependencies.success$/",
                     "group": "inline@0",
                     "icon": "$(gear)"
+                },
+                {
+                    "command": "databricks.sync.start",
+                    "when": "view == configurationView && viewItem =~ /^databricks.*sync.*is-stopped.*$/ && databricks.context.bundle.isDevTarget",
+                    "group": "inline@0"
+                },
+                {
+                    "command": "databricks.sync.stop",
+                    "when": "view == configurationView && viewItem =~ /^databricks.*sync.*is-running.*$/ && databricks.context.bundle.isDevTarget",
+                    "group": "inline@0"
                 }
             ],
             "editor/title": [

packages/databricks-vscode/src/cli/DatabricksCliSyncParser.ts

@@ -0,0 +1,150 @@
+import {logging} from "@databricks/databricks-sdk";
+import {EventEmitter} from "vscode";
+import {Loggers} from "../logger";
+import {SyncState} from "../sync";
+
+type EventBase = {
+    timestamp: string;
+    seq: number;
+    type: string;
+};
+
+type EventChanges = {
+    put: Array<string>;
+    delete: Array<string>;
+};
+
+type EventStart = EventBase &
+    EventChanges & {
+        type: "start";
+    };
+
+type EventComplete = EventBase &
+    EventChanges & {
+        type: "complete";
+    };
+
+type EventProgress = EventBase & {
+    type: "progress";
+
+    action: "put" | "delete";
+    path: string;
+    progress: number;
+};
+
+type Event = EventStart | EventComplete | EventProgress;
+
+export class DatabricksCliSyncParser {
+    private state: SyncState = "STOPPED";
+
+    constructor(
+        private syncStateCallback: (state: SyncState, reason?: string) => void,
+        private writeEmitter: EventEmitter<string>
+    ) {}
+
+    private changeSize(ec: EventChanges): number {
+        let size = 0;
+        if (ec.put) {
+            size += ec.put.length;
+        }
+        if (ec.delete) {
+            size += ec.delete.length;
+        }
+        return size;
+    }
+
+    private processLine(line: string) {
+        const event = JSON.parse(line) as Event;
+        switch (event.type) {
+            case "start": {
+                this.state = "IN_PROGRESS";
+                this.writeEmitter.fire(
+                    "Starting synchronization (" +
+                        this.changeSize(event) +
+                        " files)\r\n"
+                );
+                break;
+            }
+            case "progress": {
+                let action = "";
+                switch (event.action) {
+                    case "put":
+                        action = "Uploaded";
+                        break;
+                    case "delete":
+                        action = "Deleted";
+                        break;
+                }
+                if (event.progress === 1.0) {
+                    this.writeEmitter.fire(action + " " + event.path + "\r\n");
+                }
+                break;
+            }
+            case "complete":
+                this.state = "WATCHING_FOR_CHANGES";
+                this.writeEmitter.fire("Completed synchronization\r\n");
+                break;
+        }
+    }
+
+    public processStderr(data: string) {
+        const logLines = data.split("\n");
+        for (let i = 0; i < logLines.length; i++) {
+            const line = logLines[i].trim();
+            if (line.length === 0) {
+                continue;
+            }
+            this.writeEmitter.fire(line.trim() + "\r\n");
+            if (this.matchForErrors(line)) {
+                return;
+            }
+        }
+    }
+
+    private matchForErrors(line: string) {
+        if (line.match(/^Error: .*Files in Workspace is disabled.*/) !== null) {
+            this.syncStateCallback("FILES_IN_WORKSPACE_DISABLED");
+            return true;
+        }
+
+        if (line.match(/^Error: .*Files in Repos is disabled.*/) !== null) {
+            this.syncStateCallback("FILES_IN_REPOS_DISABLED");
+            return true;
+        }
+
+        const match = line.match(/^Error: (.*)/);
+        if (match !== null) {
+            this.syncStateCallback("ERROR", match[1]);
+            return true;
+        }
+
+        return false;
+    }
+
+    // This function processes the JSON output from databricks sync and parses it
+    // to figure out if a synchronization step is in progress or has completed.
+    public processStdout(data: string) {
+        const logLines = data.split("\n");
+        for (let i = 0; i < logLines.length; i++) {
+            const line = logLines[i].trim();
+            if (line.length === 0) {
+                continue;
+            }
+
+            try {
+                this.processLine(line);
+            } catch (error: any) {
+                logging.NamedLogger.getOrCreate(Loggers.Extension).error(
+                    "Error parsing JSON line from databricks sync stdout: " +
+                        error
+                );
+            }
+
+            if (this.matchForErrors(line)) {
+                return;
+            }
+        }
+
+        this.syncStateCallback(this.state);
+    }
+}

packages/databricks-vscode/src/cli/SyncParser.test.ts

@@ -0,0 +1,87 @@
+import "@databricks/databricks-sdk";
+import assert from "assert";
+import {mock, instance, capture, reset} from "ts-mockito";
+import {EventEmitter} from "vscode";
+import {SyncState} from "../sync";
+import {DatabricksCliSyncParser} from "./DatabricksCliSyncParser";
+
+describe("tests for SycnParser", () => {
+    let syncState: SyncState = "STOPPED";
+    let mockedOutput: EventEmitter<string>;
+    let databricksSyncParser: DatabricksCliSyncParser;
+
+    const syncStateCallback = (state: SyncState) => {
+        syncState = state;
+    };
+
+    beforeEach(() => {
+        syncState = "STOPPED";
+        mockedOutput = mock(EventEmitter<string>);
+        databricksSyncParser = new DatabricksCliSyncParser(
+            syncStateCallback,
+            instance(mockedOutput)
+        );
+    });
+
+    it("ignores empty lines", () => {
+        assert.equal(syncState, "STOPPED");
+        databricksSyncParser.processStdout("\n\n");
+        assert.equal(syncState, "STOPPED");
+    });
+
+    it("ignores non-JSON lines", () => {
+        assert.equal(syncState, "STOPPED");
+        databricksSyncParser.processStdout("foo\nbar\n");
+        assert.equal(syncState, "STOPPED");
+    });
+
+    it("transitions from STOPPED -> IN_PROGRESS -> WATCHING_FOR_CHANGES", () => {
+        assert.equal(syncState, "STOPPED");
+        databricksSyncParser.processStdout(`{"type": "start"}`);
+        assert.equal(syncState, "IN_PROGRESS");
+        databricksSyncParser.processStdout(`{"type": "complete"}`);
+        assert.equal(syncState, "WATCHING_FOR_CHANGES");
+    });
+
+    it("writes start events to the terminal", () => {
+        databricksSyncParser.processStdout(
+            `{"type": "start", "put": ["hello"], "delete": ["world"]}`
+        );
+        const arg = capture(mockedOutput.fire).first();
+        assert.match(arg[0], /^Starting synchronization /);
+    });
+
+    it("writes progress events to the terminal", () => {
+        databricksSyncParser.processStdout(
+            `{"type": "progress", "action": "put", "path": "hello", "progress": 0.0}`
+        );
+        databricksSyncParser.processStdout(
+            `{"type": "progress", "action": "put", "path": "hello", "progress": 1.0}`
+        );
+        assert.match(
+            capture(mockedOutput.fire).first()[0],
+            /^Uploaded hello\r\n/
+        );
+        reset(mockedOutput);
+
+        databricksSyncParser.processStdout(
+            `{"type": "progress", "action": "delete", "path": "hello", "progress": 0.0}`
+        );
+        databricksSyncParser.processStdout(
+            `{"type": "progress", "action": "delete", "path": "hello", "progress": 1.0}`
+        );
+        assert.match(
+            capture(mockedOutput.fire).first()[0],
+            /^Deleted hello\r\n/
+        );
+        reset(mockedOutput);
+    });
+
+    it("writes complete events to the terminal", () => {
+        databricksSyncParser.processStdout(
+            `{"type": "complete", "put": ["hello"], "delete": ["world"]}`
+        );
+        const arg = capture(mockedOutput.fire).first();
+        assert.match(arg[0], /^Completed synchronization\r\n/);
+    });
+});