Add Advanced Settings in the property panel to allow users assign certain worker nodes for operators.

core/amber/src/main/scala/edu/uci/ics/amber/clustering/ClusterListener.scala

@@ -27,6 +27,8 @@ object ClusterListener {
   final case class GetAvailableNodeAddresses()
 
   var numWorkerNodesInCluster = 0
+
+  var currentAddresses: Set[Address] = Set.empty
 }
 
 class ClusterListener extends Actor with AmberLogging {
@@ -117,9 +119,13 @@ class ClusterListener extends Actor with AmberLogging {
       case other => //skip
     }
 
+    ClusterListener.currentAddresses = getAllAddress.toSet
     numWorkerNodesInCluster = getAllAddress.size
+
+    val addressesStr = ClusterListener.currentAddresses.map(_.toString).toSeq
+
     SessionState.getAllSessionStates.foreach { state =>
-      state.send(ClusterStatusUpdateEvent(numWorkerNodesInCluster))
+      state.send(ClusterStatusUpdateEvent(numWorkerNodesInCluster, addressesStr))
     }
 
     logger.info(

core/amber/src/main/scala/edu/uci/ics/amber/engine/architecture/common/ExecutorDeployment.scala

@@ -1,8 +1,13 @@
 package edu.uci.ics.amber.engine.architecture.common
 
-import akka.actor.{Address, Deploy}
+import akka.actor.{Address, AddressFromURIString, Deploy}
 import akka.remote.RemoteScope
-import edu.uci.ics.amber.core.workflow.{PhysicalOp, PreferController, RoundRobinPreference}
+import edu.uci.ics.amber.core.workflow.{
+  GoToSpecificNode,
+  PhysicalOp,
+  PreferController,
+  RoundRobinPreference
+}
 import edu.uci.ics.amber.engine.architecture.controller.execution.OperatorExecution
 import edu.uci.ics.amber.engine.architecture.deploysemantics.AddressInfo
 import edu.uci.ics.amber.engine.architecture.pythonworker.PythonWorkflowWorker
@@ -38,6 +43,16 @@ object ExecutorDeployment {
       val preferredAddress: Address = locationPreference match {
         case PreferController =>
           addressInfo.controllerAddress
+        case node: GoToSpecificNode =>
+          val targetAddress = AddressFromURIString(node.nodeAddr)
+          addressInfo.allAddresses.find(addr => addr == targetAddress) match {
+            case Some(address) => address
+            case None =>
+              throw new IllegalStateException(
+                s"Designated node address '${node.nodeAddr}' not found among available addresses: " +
+                  addressInfo.allAddresses.map(_.host.getOrElse("None")).mkString(", ")
+              )
+          }
         case RoundRobinPreference =>
           assert(
             addressInfo.allAddresses.nonEmpty,

core/amber/src/main/scala/edu/uci/ics/amber/engine/common/AmberRuntime.scala

@@ -8,7 +8,7 @@ import edu.uci.ics.amber.clustering.ClusterListener
 import edu.uci.ics.amber.engine.architecture.messaginglayer.DeadLetterMonitorActor
 
 import java.io.{BufferedReader, InputStreamReader}
-import java.net.URL
+import java.net.{InetAddress, URL}
 import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.duration.FiniteDuration
 
@@ -46,7 +46,9 @@ object AmberRuntime {
     try {
       val query = new URL("http://checkip.amazonaws.com")
       val in = new BufferedReader(new InputStreamReader(query.openStream()))
-      in.readLine()
+      val ip = in.readLine()
+      val localIp = InetAddress.getLocalHost().getHostAddress()
+      ip
     } catch {
       case e: Exception => throw e
     }
@@ -56,18 +58,33 @@ object AmberRuntime {
     var localIpAddress = "localhost"
     if (clusterMode) {
       localIpAddress = getNodeIpAddress
-    }
 
-    val masterConfig = ConfigFactory
-      .parseString(s"""
+      val localPrivateIdAddress = InetAddress.getLocalHost().getHostAddress()
+
+      val masterConfig = ConfigFactory
+        .parseString(s"""
+              akka.remote.artery.canonical.port = 2552
+              akka.remote.artery.canonical.hostname = $localIpAddress
+              akka.remote.artery.bind.hostname = $localPrivateIdAddress
+              akka.remote.artery.bind.port = 2552
+              akka.cluster.seed-nodes = [ "akka://Amber@$localIpAddress:2552" ]
+              """)
+        .withFallback(akkaConfig)
+        .resolve()
+      AmberConfig.masterNodeAddr = createMasterAddress(localIpAddress)
+      createAmberSystem(masterConfig)
+    } else {
+      val masterConfig = ConfigFactory
+        .parseString(s"""
         akka.remote.artery.canonical.port = 2552
         akka.remote.artery.canonical.hostname = $localIpAddress
         akka.cluster.seed-nodes = [ "akka://Amber@$localIpAddress:2552" ]
         """)
-      .withFallback(akkaConfig)
-      .resolve()
-    AmberConfig.masterNodeAddr = createMasterAddress(localIpAddress)
-    createAmberSystem(masterConfig)
+        .withFallback(akkaConfig)
+        .resolve()
+      AmberConfig.masterNodeAddr = createMasterAddress(localIpAddress)
+      createAmberSystem(masterConfig)
+    }
   }
 
   def akkaConfig: Config =
@@ -79,18 +96,35 @@ object AmberRuntime {
     val addr = mainNodeAddress.getOrElse("localhost")
     var localIpAddress = "localhost"
     if (mainNodeAddress.isDefined) {
+
       localIpAddress = getNodeIpAddress
-    }
-    val workerConfig = ConfigFactory
-      .parseString(s"""
+
+      val localPrivateIdAddress = InetAddress.getLocalHost().getHostAddress()
+
+      val workerConfig = ConfigFactory
+        .parseString(s"""
+              akka.remote.artery.canonical.hostname = $localIpAddress
+              akka.remote.artery.canonical.port = 0
+              akka.remote.artery.bind.hostname = $localPrivateIdAddress
+              akka.remote.artery.bind.port = 0
+              akka.cluster.seed-nodes = [ "akka://Amber@$addr:2552" ]
+              """)
+        .withFallback(akkaConfig)
+        .resolve()
+      AmberConfig.masterNodeAddr = createMasterAddress(addr)
+      createAmberSystem(workerConfig)
+    } else {
+      val workerConfig = ConfigFactory
+        .parseString(s"""
         akka.remote.artery.canonical.hostname = $localIpAddress
         akka.remote.artery.canonical.port = 0
         akka.cluster.seed-nodes = [ "akka://Amber@$addr:2552" ]
         """)
-      .withFallback(akkaConfig)
-      .resolve()
-    AmberConfig.masterNodeAddr = createMasterAddress(addr)
-    createAmberSystem(workerConfig)
+        .withFallback(akkaConfig)
+        .resolve()
+      AmberConfig.masterNodeAddr = createMasterAddress(addr)
+      createAmberSystem(workerConfig)
+    }
   }
 
   private def createAmberSystem(actorSystemConf: Config): Unit = {

core/amber/src/main/scala/edu/uci/ics/texera/web/model/websocket/response/ClusterStatusUpdateEvent.scala

@@ -2,4 +2,5 @@ package edu.uci.ics.texera.web.model.websocket.response
 
 import edu.uci.ics.texera.web.model.websocket.event.TexeraWebSocketEvent
 
-case class ClusterStatusUpdateEvent(numWorkers: Int) extends TexeraWebSocketEvent
+case class ClusterStatusUpdateEvent(numWorkers: Int, addresses: Seq[String] = Seq.empty)
+    extends TexeraWebSocketEvent

core/amber/src/main/scala/edu/uci/ics/texera/web/resource/WorkflowWebsocketResource.scala

@@ -36,7 +36,10 @@ class WorkflowWebsocketResource extends LazyLogging {
     val workflowState =
       WorkflowService.getOrCreate(WorkflowIdentity(wid))
     sessionState.subscribe(workflowState)
-    sessionState.send(ClusterStatusUpdateEvent(ClusterListener.numWorkerNodesInCluster))
+    val addressesStr = ClusterListener.currentAddresses.map(_.toString).toSeq
+    sessionState.send(
+      ClusterStatusUpdateEvent(ClusterListener.numWorkerNodesInCluster, addressesStr)
+    )
     logger.info("connection open")
   }
 

core/gui/src/app/app.module.ts

@@ -81,6 +81,7 @@ import { AdminGuardService } from "./dashboard/service/admin/guard/admin-guard.s
 import { ContextMenuComponent } from "./workspace/component/workflow-editor/context-menu/context-menu/context-menu.component";
 import { CoeditorUserIconComponent } from "./workspace/component/menu/coeditor-user-icon/coeditor-user-icon.component";
 import { InputAutoCompleteComponent } from "./workspace/component/input-autocomplete/input-autocomplete.component";
+import { InputNodeAddressComponent } from "./workspace/component/input-node-address/input-node-address.component";
 import { CollabWrapperComponent } from "./common/formly/collab-wrapper/collab-wrapper/collab-wrapper.component";
 import { NzSwitchModule } from "ng-zorro-antd/switch";
 import { AboutComponent } from "./hub/component/about/about.component";
@@ -214,6 +215,7 @@ registerLocaleData(en);
     ContextMenuComponent,
     CoeditorUserIconComponent,
     InputAutoCompleteComponent,
+    InputNodeAddressComponent,
     FileSelectionComponent,
     CollabWrapperComponent,
     AboutComponent,

core/gui/src/app/common/formly/formly-config.ts

@@ -7,6 +7,7 @@ import { CodeareaCustomTemplateComponent } from "../../workspace/component/codea
 import { PresetWrapperComponent } from "./preset-wrapper/preset-wrapper.component";
 import { InputAutoCompleteComponent } from "../../workspace/component/input-autocomplete/input-autocomplete.component";
 import { CollabWrapperComponent } from "./collab-wrapper/collab-wrapper/collab-wrapper.component";
+import { InputNodeAddressComponent } from "../../workspace/component/input-node-address/input-node-address.component";
 
 /**
  * Configuration for using Json Schema with Formly.
@@ -57,6 +58,7 @@ export const TEXERA_FORMLY_CONFIG = {
     { name: "multischema", component: MultiSchemaTypeComponent },
     { name: "codearea", component: CodeareaCustomTemplateComponent },
     { name: "inputautocomplete", component: InputAutoCompleteComponent, wrappers: ["form-field"] },
+    { name: "inputnodeaddress", component: InputNodeAddressComponent, wrappers: ["form-field"] },
   ],
   wrappers: [
     { name: "preset-wrapper", component: PresetWrapperComponent },

core/gui/src/app/workspace/component/input-node-address/input-node-address.component.html

@@ -0,0 +1,10 @@
+<nz-select
+  [formControl]="formControl"
+  [nzPlaceHolder]="'Select Node Address'"
+  nzAllowClear>
+  <nz-option
+    *ngFor="let addr of nodeAddresses"
+    [nzValue]="addr"
+    [nzLabel]="addr">
+  </nz-option>
+</nz-select>

core/gui/src/app/workspace/component/input-node-address/input-node-address.component.scss

@@ -0,0 +1,7 @@
+:host {
+  display: block;
+}
+
+nz-select {
+  width: 100%;
+}

core/gui/src/app/workspace/component/input-node-address/input-node-address.component.ts

@@ -0,0 +1,40 @@
+import { Component, OnInit, OnDestroy } from "@angular/core";
+import { FieldType, FieldTypeConfig } from "@ngx-formly/core";
+import { Subject, Subscription } from "rxjs";
+import { WorkflowWebsocketService } from "../../service/workflow-websocket/workflow-websocket.service";
+import { takeUntil } from "rxjs/operators";
+
+@Component({
+  selector: "texera-input-node-address",
+  templateUrl: "./input-node-address.component.html",
+  styleUrls: ["./input-node-address.component.scss"],
+})
+export class InputNodeAddressComponent extends FieldType<FieldTypeConfig> implements OnInit, OnDestroy {
+  public nodeAddresses: string[] = [];
+
+  private wsSubscription!: Subscription;
+
+  private componentDestroy = new Subject<void>();
+
+  constructor(private workflowWebsocketService: WorkflowWebsocketService) {
+    super();
+  }
+
+  ngOnInit(): void {
+    this.nodeAddresses = this.workflowWebsocketService.workerAddresses;
+
+    this.wsSubscription = this.workflowWebsocketService
+      .websocketEvent()
+      .pipe(takeUntil(this.componentDestroy))
+      .subscribe(event => {
+        if (event.type === "ClusterStatusUpdateEvent") {
+          this.nodeAddresses = event.addresses;
+        }
+      });
+  }
+
+  ngOnDestroy(): void {
+    this.componentDestroy.next();
+    this.componentDestroy.complete();
+  }
+}

core/gui/src/app/workspace/component/property-editor/operator-property-edit-frame/operator-property-edit-frame.component.html

@@ -69,23 +69,53 @@
   (focusout)="disconnectQuillFromText()"
   (keyup.enter)="disconnectQuillFromText()"></div>
 
-<form
-  nz-form
-  [nzLayout]="'vertical'"
-  *ngIf="formlyFields && formlyFormGroup"
-  [formGroup]="formlyFormGroup"
-  class="property-editor-form">
-  <formly-form
-    (modelChange)="onFormChanges($event)"
-    [fields]="formlyFields"
-    [form]="formlyFormGroup"
-    [model]="formData"
-    [options]="formlyOptions">
-  </formly-form>
-  <texera-type-casting-display
-    *ngIf="isTypeCasting"
-    currentOperatorId="{{this.currentOperatorId}}"></texera-type-casting-display>
-</form>
+<nz-tabset>
+  <nz-tab
+    nzTitle="Basic Settings"
+    *ngIf="basicFields.length > 0">
+    <form
+      nz-form
+      nzLayout="vertical"
+      *ngIf="formlyFields && formlyFormGroup"
+      [formGroup]="formlyFormGroup"
+      class="property-editor-form">
+      <formly-form
+        (modelChange)="onFormChanges($event)"
+        [fields]="basicFields"
+        [form]="formlyFormGroup"
+        [model]="formData"
+        [options]="formlyOptions">
+      </formly-form>
+
+      <texera-type-casting-display
+        *ngIf="isTypeCasting"
+        currentOperatorId="{{this.currentOperatorId}}"></texera-type-casting-display>
+    </form>
+  </nz-tab>
+
+  <nz-tab
+    nzTitle="Advanced Settings"
+    *ngIf="advancedFields.length > 0">
+    <form
+      nz-form
+      nzLayout="vertical"
+      *ngIf="formlyFields && formlyFormGroup"
+      [formGroup]="formlyFormGroup"
+      class="property-editor-form">
+      <formly-form
+        (modelChange)="onFormChanges($event)"
+        [fields]="advancedFields"
+        [form]="formlyFormGroup"
+        [model]="formData"
+        [options]="formlyOptions">
+      </formly-form>
+
+      <texera-type-casting-display
+        *ngIf="isTypeCasting"
+        currentOperatorId="{{this.currentOperatorId}}"></texera-type-casting-display>
+    </form>
+  </nz-tab>
+</nz-tabset>
 
 <button
   (click)="allowModifyOperatorLogic()"

core/gui/src/app/workspace/component/property-editor/operator-property-edit-frame/operator-property-edit-frame.component.ts

@@ -104,6 +104,9 @@ export class OperatorPropertyEditFrameComponent implements OnInit, OnChanges, On
   formlyFields: FormlyFieldConfig[] | undefined;
   formTitle: string | undefined;
 
+  public basicFields: FormlyFieldConfig[] = [];
+  public advancedFields: FormlyFieldConfig[] = [];
+
   // The field name and its css style to be overridden, e.g., for showing the diff between two workflows.
   // example: new Map([
   //     ["attribute", "outline: 3px solid green; transition: 0.3s ease-in-out outline;"],
@@ -379,6 +382,20 @@ export class OperatorPropertyEditFrameComponent implements OnInit, OnChanges, On
         },
       };
 
+      if (mappedField.key == "nodeAddr") {
+        mappedField.props = {
+          ...mappedField.props,
+          tab: "AdvancedSettings",
+        };
+      }
+
+      if (mappedField.key == "autoSelectNodeAddress") {
+        mappedField.props = {
+          ...mappedField.props,
+          tab: "AdvancedSettings",
+        };
+      }
+
       // Disable dummy operator for user
       if (mappedField.key === "dummyOperator") {
         mappedField.expressions = {
@@ -432,6 +449,11 @@ export class OperatorPropertyEditFrameComponent implements OnInit, OnChanges, On
         mappedField.type = "inputautocomplete";
       }
 
+      // if the title is nodeAddr, then change it to custom inputnodeaddress input template
+      if (mappedField.key == "nodeAddr") {
+        mappedField.type = "inputnodeaddress";
+      }
+
       // if the title is python script (for Python UDF), then make this field a custom template 'codearea'
       if (mapSource?.description?.toLowerCase() === "input your code here") {
         if (mappedField.type) {
@@ -669,6 +691,10 @@ export class OperatorPropertyEditFrameComponent implements OnInit, OnChanges, On
     // not return field.fieldGroup directly because
     // doing so the validator in the field will not be triggered
     this.formlyFields = [field];
+
+    const fieldGroup = this.formlyFields?.[0]?.fieldGroup ?? [];
+    this.basicFields = fieldGroup.filter(f => f.props?.tab !== "AdvancedSettings");
+    this.advancedFields = fieldGroup.filter(f => f.props?.tab === "AdvancedSettings");
   }
 
   allowModifyOperatorLogic(): void {