Add telemetry event for SSH tunnel connections (#4881)

## Changes

Add a `SshTunnelEvent` telemetry event that fires once per `ssh connect`
/ `ssh proxy` / IDE-mode invocation in
`experimental/ssh/internal/client/client.go`.

Captured fields:
- `compute_type` — dedicated cluster or serverless
- `accelerator_type` — GPU accelerator for serverless (empty for
dedicated)
- `ide_type` — IDE that initiated the connection (e.g. `vscode`,
`cursor`), empty for raw SSH / proxy
- `client_mode` — `SSH_CLIENT`, `PROXY`, or `IDE`
- `is_reconnect` — true when invoked with `ServerMetadata` (a follow-up
connection re-using an existing server)
- `auto_start_cluster` — whether `--auto-start-cluster` was set
- `server_start_time_ms` — time spent inside `ensureSSHServerIsRunning`;
0 on reconnect
- `is_success` — set to true only after the connection is fully
established

New files:
- `libs/telemetry/protos/ssh_tunnel.go` — event struct and the
`SshTunnelComputeType` / `SshTunnelClientMode` string-enum types,
following the same pattern as `BundleMode` in `enum.go`.
- Wires the event into `DatabricksCliLog` in
`libs/telemetry/protos/frontend_log.go`.

Emission is done from a `defer` registered before the first early-return
path inside `Run`, so the event fires on every exit (cluster-state
failure, binary-upload failure, metadata-parse failure, success) rather
than only the original two explicit call sites.

## Why

We have no visibility into how the experimental SSH tunnel feature is
being used. We want to answer basic questions before promoting it out of
experimental:
- Dedicated vs serverless split
- IDE breakdown (raw SSH client vs proxy vs IDE-initiated, and which
IDEs)
- Success rate and where failures cluster
- How long server startup takes on serverless (cold start signal)

The `defer` pattern is deliberate: an event emitted only on the happy
path would under-report failures and bias all of the above metrics.

## Tests

- `task test-exp-ssh` passes on linux/macos/windows (all green in CI).
- `task lint` clean.
- Verified locally that the event is recorded on the in-memory telemetry
logger for both dedicated and serverless invocations, including failure
paths (cluster-state check failure, binary upload failure).

No new test files: this repo has no existing tests asserting on
telemetry event payloads — `BundleInitEvent` and `BundleDeployEvent` are
emitted untested as well (`bundle/phases/telemetry.go`,
`libs/template/writer.go`). Adding a fixture-based test only for
`SshTunnelEvent` would be inconsistent with the rest of the codebase;
that work should land as a separate PR if we want it.

No `NEXT_CHANGELOG.md` entry: the feature is under `experimental/` and
the change is internal telemetry, not a user-visible behavior change.

Author: anton-107

experimental/ssh/internal/client/client.go

@@ -29,6 +29,8 @@ import (
 	"github.com/databricks/cli/libs/auth"
 	"github.com/databricks/cli/libs/cmdio"
 	"github.com/databricks/cli/libs/log"
+	"github.com/databricks/cli/libs/telemetry"
+	"github.com/databricks/cli/libs/telemetry/protos"
 	"github.com/databricks/databricks-sdk-go"
 	"github.com/databricks/databricks-sdk-go/retries"
 	"github.com/databricks/databricks-sdk-go/service/compute"
@@ -265,6 +267,13 @@ func Run(ctx context.Context, client *databricks.WorkspaceClient, opts ClientOpt
 		}
 	}
 
+	isReconnect := opts.ServerMetadata != ""
+	var serverStartTimeMs int64
+	isSuccess := false
+	defer func() {
+		logSshTunnelEvent(ctx, opts, isSuccess, isReconnect, serverStartTimeMs)
+	}()
+
 	// Only check cluster state for dedicated clusters
 	if !opts.IsServerlessMode() {
 		cmdio.LogString(ctx, "Checking cluster state...")
@@ -311,6 +320,7 @@ func Run(ctx context.Context, client *databricks.WorkspaceClient, opts ClientOpt
 		if err != nil {
 			return fmt.Errorf("failed to upload ssh-tunnel binaries: %w", err)
 		}
+		serverStartTime := time.Now()
 		userName, serverPort, clusterID, err = ensureSSHServerIsRunning(ctx, client, version, secretScopeName, opts)
 		if err != nil {
 			if opts.IsServerlessMode() && opts.Accelerator == "" && errors.Is(err, errServerMetadata) {
@@ -319,6 +329,7 @@ func Run(ctx context.Context, client *databricks.WorkspaceClient, opts ClientOpt
 			}
 			return fmt.Errorf("failed to ensure that ssh server is running: %w", err)
 		}
+		serverStartTimeMs = time.Since(serverStartTime).Milliseconds()
 	} else {
 		// Metadata format: "<user_name>,<port>,<cluster_id>"
 		metadata := strings.Split(opts.ServerMetadata, ",")
@@ -355,6 +366,8 @@ func Run(ctx context.Context, client *databricks.WorkspaceClient, opts ClientOpt
 		cmdio.LogString(ctx, "Connected!")
 	}
 
+	isSuccess = true
+
 	if opts.ProxyMode {
 		return runSSHProxy(ctx, client, serverPort, clusterID, opts)
 	} else if opts.IDE != "" {
@@ -728,3 +741,33 @@ func ensureSSHServerIsRunning(ctx context.Context, client *databricks.WorkspaceC
 
 	return userName, serverPort, effectiveClusterID, nil
 }
+
+func logSshTunnelEvent(ctx context.Context, opts ClientOptions, isSuccess, isReconnect bool, serverStartTimeMs int64) {
+	computeType := protos.SshTunnelComputeTypeDedicated
+	if opts.IsServerlessMode() {
+		computeType = protos.SshTunnelComputeTypeServerless
+	}
+
+	var clientMode protos.SshTunnelClientMode
+	switch {
+	case opts.ProxyMode:
+		clientMode = protos.SshTunnelClientModeProxy
+	case opts.IDE != "":
+		clientMode = protos.SshTunnelClientModeIDE
+	default:
+		clientMode = protos.SshTunnelClientModeSSH
+	}
+
+	telemetry.Log(ctx, protos.DatabricksCliLog{
+		SshTunnelEvent: &protos.SshTunnelEvent{
+			ComputeType:       computeType,
+			AcceleratorType:   opts.Accelerator,
+			IdeType:           opts.IDE,
+			ClientMode:        clientMode,
+			IsReconnect:       isReconnect,
+			AutoStartCluster:  opts.AutoStartCluster,
+			ServerStartTimeMs: serverStartTimeMs,
+			IsSuccess:         isSuccess,
+		},
+	})
+}

libs/telemetry/protos/frontend_log.go

@@ -19,4 +19,5 @@ type DatabricksCliLog struct {
 	CliTestEvent      *CliTestEvent      `json:"cli_test_event,omitempty"`
 	BundleInitEvent   *BundleInitEvent   `json:"bundle_init_event,omitempty"`
 	BundleDeployEvent *BundleDeployEvent `json:"bundle_deploy_event,omitempty"`
+	SshTunnelEvent    *SshTunnelEvent    `json:"ssh_tunnel_event,omitempty"`
 }

libs/telemetry/protos/ssh_tunnel.go

@@ -0,0 +1,47 @@
+package protos
+
+type SshTunnelComputeType string
+
+const (
+	SshTunnelComputeTypeUnspecified SshTunnelComputeType = "TYPE_UNSPECIFIED"
+	SshTunnelComputeTypeDedicated   SshTunnelComputeType = "DEDICATED"
+	SshTunnelComputeTypeServerless  SshTunnelComputeType = "SERVERLESS"
+)
+
+type SshTunnelClientMode string
+
+const (
+	SshTunnelClientModeUnspecified SshTunnelClientMode = "TYPE_UNSPECIFIED"
+	SshTunnelClientModeSSH         SshTunnelClientMode = "SSH_CLIENT"
+	SshTunnelClientModeProxy       SshTunnelClientMode = "PROXY"
+	SshTunnelClientModeIDE         SshTunnelClientMode = "IDE"
+)
+
+// SshTunnelEvent is emitted when a user establishes an SSH tunnel connection
+// via the Databricks CLI.
+type SshTunnelEvent struct {
+	// Type of compute: dedicated cluster or serverless.
+	ComputeType SshTunnelComputeType `json:"compute_type,omitempty"`
+
+	// GPU accelerator type for serverless compute.
+	AcceleratorType string `json:"accelerator_type,omitempty"`
+
+	// IDE that initiated the connection (e.g., "vscode", "cursor").
+	IdeType string `json:"ide_type,omitempty"`
+
+	// How the client is used: SSH client, proxy mode, or IDE mode.
+	ClientMode SshTunnelClientMode `json:"client_mode,omitempty"`
+
+	// Whether this is a reconnection to an existing session.
+	IsReconnect bool `json:"is_reconnect,omitempty"`
+
+	// Whether the cluster was auto-started by the CLI.
+	AutoStartCluster bool `json:"auto_start_cluster,omitempty"`
+
+	// Time in milliseconds spent starting the SSH server.
+	// Zero if server was already running.
+	ServerStartTimeMs int64 `json:"server_start_time_ms"`
+
+	// Whether the connection was successful.
+	IsSuccess bool `json:"is_success,omitempty"`
+}