diff --git a/packages/business/src/components/error-message/index.tsx b/packages/business/src/components/error-message/index.tsx index 9cb3e7f61a..69e43a1b4a 100644 --- a/packages/business/src/components/error-message/index.tsx +++ b/packages/business/src/components/error-message/index.tsx @@ -3,6 +3,10 @@ import i18n from '@tap/i18n' import { copyToClipboard } from '@tap/shared' import './index.scss' +let lastErrorMessage = '' +let lastErrorAt = 0 +const DUPLICATE_ERROR_WINDOW_MS = 1200 + function renderDialog(stack: string) { return (
@@ -103,10 +107,20 @@ export function ErrorMessage(stack: string, message: string) { } export function showErrorMessage(error: Record) { - let message = - error?.message || - error?.msg || + const response = error?.response?.data ?? error + const detail = + response?.code === 'Task.ListWarnMessage' && response.data + ? Object.values(response.data) + .flatMap((items) => (Array.isArray(items) ? items : [items])) + .find((item: any) => item?.msg || item?.message) + : undefined + const rawMessage = + detail?.msg || + detail?.message || + response?.message || + response?.msg || i18n.global.t('public_message_request_error') + let message = rawMessage let duration = 3000 let showClose = false @@ -123,6 +137,16 @@ export function showErrorMessage(error: Record) { showClose = true } + const now = Date.now() + if ( + rawMessage === lastErrorMessage && + now - lastErrorAt < DUPLICATE_ERROR_WINDOW_MS + ) { + return + } + lastErrorMessage = rawMessage + lastErrorAt = now + ElMessage.error({ message, duration, diff --git a/packages/business/src/views/heartbeat-table/List.vue b/packages/business/src/views/heartbeat-table/List.vue index 4729b82a12..4eddeff4ef 100644 --- a/packages/business/src/views/heartbeat-table/List.vue +++ b/packages/business/src/views/heartbeat-table/List.vue @@ -10,6 +10,7 @@ import { } from '@tap/api/src/core/task' import { requestClient } from '@tap/api/src/request' import { FilterBar } from '@tap/component/src/filter-bar' +import { showErrorMessage } from '../../components/error-message' import { useI18n } from '@tap/i18n' import dayjs from 'dayjs' import { ElMessage, ElMessageBox } from 'element-plus' @@ -178,14 +179,37 @@ const start = (ids: string[]) => { fetchTasks({ where: { id: ids[0] } }).then(() => { batchStartTasks(ids) .then((data: any) => { + const resultList = Array.isArray(data) + ? data + : Array.isArray(data?.data) + ? data.data + : [] + const failed = resultList.filter( + (item) => item?.code && item.code !== 'ok', + ) + if (failed.length) { + showErrorMessage({ + msg: failed + .map((item) => item.message) + .filter(Boolean) + .join('\n'), + }) + table.value?.fetch() + buried?.(taskBuried.start, '', { result: false }) + return + } buried?.(taskBuried.start, '', { result: true }) ElMessage.success( data?.message || t('public_message_operation_success'), ) table.value?.fetch() }) - .catch(() => { + .catch((error) => { buried?.(taskBuried.start, '', { result: false }) + showErrorMessage({ + msg: error?.message || t('public_message_operation_failed'), + stack: error?.stack, + }) }) }) } diff --git a/packages/business/src/views/shared-cache/List.vue b/packages/business/src/views/shared-cache/List.vue index f9da607596..38cc2cbb90 100644 --- a/packages/business/src/views/shared-cache/List.vue +++ b/packages/business/src/views/shared-cache/List.vue @@ -15,6 +15,7 @@ import { } from '@tap/api/src/core/task' import { requestClient, withPassive } from '@tap/api/src/request' import { FilterBar } from '@tap/component/src/filter-bar' +import { showErrorMessage } from '../../components/error-message' import { useI18n } from '@tap/i18n' import dayjs from 'dayjs' import { ElMessage, ElMessageBox } from 'element-plus' @@ -238,14 +239,37 @@ const start = async (ids: string[], row: any) => { fetchTasks({ where: { id: ids[0] } }).then(() => { batchStartTasks(ids) .then((data: any) => { + const resultList = Array.isArray(data) + ? data + : Array.isArray(data?.data) + ? data.data + : [] + const failed = resultList.filter( + (item) => item?.code && item.code !== 'ok', + ) + if (failed.length) { + showErrorMessage({ + msg: failed + .map((item) => item.message) + .filter(Boolean) + .join('\n'), + }) + table.value?.fetch() + buried?.(taskBuried.start, '', { result: false }) + return + } buried?.(taskBuried.start, '', { result: true }) ElMessage.success( data?.message || t('public_message_operation_success'), ) table.value?.fetch() }) - .catch(() => { + .catch((error) => { buried?.(taskBuried.start, '', { result: false }) + showErrorMessage({ + msg: error?.message || t('public_message_operation_failed'), + stack: error?.stack, + }) }) }) } diff --git a/packages/business/src/views/shared-mining/List.vue b/packages/business/src/views/shared-mining/List.vue index 640a8f514e..86dc8f6474 100644 --- a/packages/business/src/views/shared-mining/List.vue +++ b/packages/business/src/views/shared-mining/List.vue @@ -18,6 +18,7 @@ import { import { requestClient, withPassive } from '@tap/api/src/request' import { VTable } from '@tap/component/src/base/v-table' import { FilterBar } from '@tap/component/src/filter-bar' +import { showErrorMessage } from '../../components/error-message' import { useI18n } from '@tap/i18n' import { calcTimeUnit, openUrl } from '@tap/shared' import dayjs from 'dayjs' @@ -276,14 +277,37 @@ const start = (ids: string[]) => { fetchTasks({ where: { id: ids[0] } }).then(() => { batchStartTasks(ids) .then((data: any) => { + const resultList = Array.isArray(data) + ? data + : Array.isArray(data?.data) + ? data.data + : [] + const failed = resultList.filter( + (item) => item?.code && item.code !== 'ok', + ) + if (failed.length) { + showErrorMessage({ + msg: failed + .map((item) => item.message) + .filter(Boolean) + .join('\n'), + }) + table.value?.fetch() + buried?.(taskBuried.start, '', { result: false }) + return + } buried?.(taskBuried.start, '', { result: true }) ElMessage.success( data?.message || t('public_message_operation_success'), ) table.value?.fetch() }) - .catch(() => { + .catch((error) => { buried?.(taskBuried.start, '', { result: false }) + showErrorMessage({ + msg: error?.message || t('public_message_operation_failed'), + stack: error?.stack, + }) }) }) } diff --git a/packages/business/src/views/task/List.vue b/packages/business/src/views/task/List.vue index 41b5ee4e3a..952c129d51 100644 --- a/packages/business/src/views/task/List.vue +++ b/packages/business/src/views/task/List.vue @@ -564,7 +564,12 @@ export default { }, responseHandler(data, msg, canNotList = []) { - let failList = data?.filter((t) => t.code !== 'ok') || [] + const resultList = Array.isArray(data) + ? data + : Array.isArray(data?.data) + ? data.data + : [] + let failList = resultList.filter((t) => t.code !== 'ok') failList = [...failList, ...canNotList] if (failList.length) { const findErr = failList.find((t) => @@ -706,8 +711,12 @@ export default { canNotList, ) }) - .catch(() => { + .catch((error) => { this.buried(this.taskBuried.start, '', { result: false }) + showErrorMessage({ + msg: error?.message || this.$t('public_message_operation_failed'), + stack: error?.stack, + }) }) }, diff --git a/packages/dag/src/MigrationMonitorSimple.vue b/packages/dag/src/MigrationMonitorSimple.vue index efda7dbd3a..686206ec0a 100644 --- a/packages/dag/src/MigrationMonitorSimple.vue +++ b/packages/dag/src/MigrationMonitorSimple.vue @@ -968,7 +968,8 @@ export default { isError = true } }) - isError && (someErrorMsg = this.$t('packages_dag_mixins_editor_suoshuage')) + isError && + (someErrorMsg = this.$t('packages_dag_mixins_editor_suoshuage')) } } else if (accessNodeProcessIdArr.length === 1) { // 如果画布上仅有一个所属agent，自动设置为任务的agent @@ -1049,7 +1050,18 @@ export default { this.isSaving = true try { this.wsAgentLive() - await startTask(this.dataflow.id) + const response = await startTask(this.dataflow.id) + const payload = response?.data ?? response + const businessError = Array.isArray(payload) + ? payload.find((item) => item?.code && item.code !== 'ok') + : payload?.code && payload.code !== 'ok' + ? payload + : null + if (businessError) { + this.handleError(businessError) + this.isSaving = false + return + } this.$message.success(this.$t('public_message_operation_success')) this.isSaving = false this.isReset = false diff --git a/packages/dag/src/components/LeftSidebar.vue b/packages/dag/src/components/LeftSidebar.vue index 6bdef1330a..2de5a2a23d 100644 --- a/packages/dag/src/components/LeftSidebar.vue +++ b/packages/dag/src/components/LeftSidebar.vue @@ -524,6 +524,7 @@ export default { connectionName: connection.name, connectionType: connection.connection_type, accessNodeProcessId: connection.accessNodeProcessId, + accessNodeProcessIdList: connection.accessNodeProcessIdList || [], priorityProcessId: connection.priorityProcessId, pdkType: connection.pdkType, pdkHash: connection.pdkHash, diff --git a/packages/dag/src/components/SettingsPanel.vue b/packages/dag/src/components/SettingsPanel.vue index c4a304a347..cf012c39c8 100644 --- a/packages/dag/src/components/SettingsPanel.vue +++ b/packages/dag/src/components/SettingsPanel.vue @@ -43,6 +43,8 @@ import { getDefaultAlarmSettings, type AlarmSettingKey, } from '../constants' +import { useConnectionAgentConfigs } from '../composables/useConnectionAgentConfigs' +import { getCompatibleAccessNodeOptions } from '../utils/accessNodePolicy' import { useDataflowStore } from '../stores/dataflow.store' const dataflowStore = useDataflowStore() @@ -129,8 +131,10 @@ interface FormScope { ) => (field: any) => void loadRoleList: (field: any, val: any[]) => Promise handleRemovePermissionsItem: () => void - getConnectionNameByAgent: (field: any) => void handleQuicklySyncPoints: () => void + $getTaskAgentOptions: (accessNodeType: string) => any[] + $getTaskAgentConnectionCount: () => number + $refreshTaskAgentConfigs: () => Promise } // Store @@ -508,6 +512,14 @@ const formScope: FormScope = { }, getPickerOptionsBeforeTime, $isDaas: isDaas, + $getTaskAgentOptions: (accessNodeType: string) => + getCompatibleAccessNodeOptions( + scope.$agents || [], + connectionAgentConfigList.value, + accessNodeType, + ), + $getTaskAgentConnectionCount: () => connectionIds.value.length, + $refreshTaskAgentConfigs: () => refreshConnectionAgentConfigs(), formTab: FormTab.createFormTab(openAlarmTab ? 'tab3' : undefined), checkName: (value: string) => { return new Promise((resolve) => { @@ -573,31 +585,6 @@ const formScope: FormScope = { handleRemovePermissionsItem: () => { savePermissionsConfig() }, - getConnectionNameByAgent: (field: any) => { - if ( - !field.value || - !field.dataSource?.length || - !accessNodeProcessIdMap.value[field.value] - ) { - field.setDescription('') - return - } - - const map: Record = {} - - for (const id of accessNodeProcessIdMap.value[field.value]) { - const node = dataflowStore.findNodeById(id) - map[node.connectionId] = node.attrs.connectionName - } - - const values = Object.values(map) - - field.setDescription( - values.length - ? `${t('packages_dag_agent_setting_from')}: ${values.join(', ')}` - : '', - ) - }, handleQuicklySyncPoints: () => { const currentEventTimestamp = form.values.currentEventTimestamp const syncPoints = form.values.syncPoints @@ -626,35 +613,27 @@ const showDoubleActive = computed(() => { : false }) -const accessNodeProcessIdMap = computed(() => { - return dataNodes.value.reduce((map: Record, node: any) => { - const { accessNodeProcessId } = node.attrs - if (accessNodeProcessId) { - let nodeIdArr = map[accessNodeProcessId] - - if (!nodeIdArr) { - nodeIdArr = map[accessNodeProcessId] = [] - } - - nodeIdArr.push(node.id) - } - return map - }, {}) -}) +const connectionIds = computed(() => [ + ...new Set( + dataNodes.value.map((node: any) => node.connectionId).filter(Boolean), + ), +]) +const { connectionAgentConfigs, refreshConnectionAgentConfigs } = + useConnectionAgentConfigs(connectionIds) -const accessNodeProcessIdArr = computed(() => - Object.keys(accessNodeProcessIdMap.value), +watch( + () => dataflowStore.showSettings, + async (visible) => { + if (visible) await refreshConnectionAgentConfigs() + }, ) -const accessNodeProcessList = computed(() => { - const agents = scope.$agents.filter( - (item: any) => item.accessNodeType === dataflow.value.accessNodeType, - ) - if (!accessNodeProcessIdArr.value.length) return agents - return agents.filter( - (item: any) => !!accessNodeProcessIdMap.value[item.value], - ) -}) +const connectionAgentConfigList = computed(() => + dataNodes.value.map((node: any) => ({ + ...(node.attrs || {}), + ...(connectionAgentConfigs.value[node.connectionId] || {}), + })), +) const sourceNodes = computed(() => { return allNodes.value @@ -766,57 +745,6 @@ watch(stateIsReadonly, (v) => { } }) -watch( - accessNodeProcessIdArr, - (arr) => { - const size = arr.length - if (size >= 1) { - let currentId = form.values.accessNodeProcessId - currentId = currentId && arr.includes(currentId) ? currentId : arr[0] - - dataflowStore.dataflow.accessNodeType = - scope.$agentMap[currentId]?.accessNodeType || - 'MANUALLY_SPECIFIED_BY_THE_USER' - dataflowStore.dataflow.accessNodeProcessId = currentId - - if ( - form.values.accessNodeType === - 'MANUALLY_SPECIFIED_BY_THE_USER_AGENT_GROUP' - ) { - const nodeIds = accessNodeProcessIdMap.value[currentId] - let priorityProcessId = null - - nodeIds.some((id) => { - const node = scope.findNodeById(id) - if (node && node.attrs.priorityProcessId) { - priorityProcessId = node.attrs.priorityProcessId - return true - } - return false - }) - - dataflowStore.dataflow.priorityProcessId = priorityProcessId - } - } - if (!stateIsReadonly.value) { - form.setFieldState('*(accessNodeType,accessNodeProcessId)', { - disabled: size === 1, - }) - } - }, - { deep: true, immediate: true }, -) - -watch( - accessNodeProcessList, - (dataSource = []) => { - form.setFieldState('accessNodeProcessId', { - dataSource, - }) - }, - { deep: true, immediate: true }, -) - const syncPointsCache: Record = {} watch( @@ -1460,141 +1388,36 @@ const schema = { }, 'x-component': 'Switch', }, - accessNodeType: { - type: 'string', + taskAgentSettings: { + type: 'void', title: t('packages_dag_connection_form_access_node'), - default: 'AUTOMATIC_PLATFORM_ALLOCATION', 'x-decorator': 'FormItem', - 'x-component': 'Select', - enum: [ - { - label: t( - 'packages_dag_connection_form_automatic', - ), - value: 'AUTOMATIC_PLATFORM_ALLOCATION', - }, - { - label: t('packages_dag_connection_form_manual'), - value: 'MANUALLY_SPECIFIED_BY_THE_USER', - }, - { - label: t( - 'packages_business_connection_form_group', - ), - value: - 'MANUALLY_SPECIFIED_BY_THE_USER_AGENT_GROUP', - }, - ], - 'x-reactions': [ - { - fulfill: { - state: { - dataSource: `{{$isDaas ? $self.dataSource : $self.dataSource.slice(0,2)}}`, - }, - }, - }, - { - target: 'accessNodeProcessId', - effects: ['onFieldInputValueChange'], - fulfill: { - state: { - value: '', - // '{{$target.value || (item = $target.dataSource.find(item => !item.disabled), item ? item.value:undefined)}}' - }, - }, - }, - ], - }, - - agentWrap: { - type: 'void', - 'x-component': 'Space', - 'x-component-props': { - class: 'w-100 align-items-start', - }, - 'x-reactions': { - dependencies: ['.accessNodeType'], - fulfill: { - state: { - visible: - "{{['MANUALLY_SPECIFIED_BY_THE_USER', 'MANUALLY_SPECIFIED_BY_THE_USER_AGENT_GROUP'].includes($deps[0])}}", - }, - }, + 'x-decorator-props': { + feedbackLayout: 'none', }, - properties: { - accessNodeProcessId: { - type: 'string', - 'x-decorator': 'FormItem', - 'x-decorator-props': { - class: 'flex-1', - }, - 'x-component': 'Select', - 'x-reactions': [ - '{{getConnectionNameByAgent}}', - // 根据下拉数据判断是否存在已选的agent - { - dependencies: [ - '.accessNodeType', - '.accessNodeOption#dataSource', - ], - fulfill: { - state: { - title: `{{'MANUALLY_SPECIFIED_BY_THE_USER_AGENT_GROUP' === $deps[0] ? '${t( - 'packages_business_choose_agent_group', - )}': '${t( - 'packages_business_choose_agent', - )}'}}`, - }, - }, - }, - ], - }, - priorityProcessId: { - title: t('packages_business_priorityProcessId'), - type: 'string', - default: '', - 'x-decorator': 'FormItem', - 'x-decorator-props': { - class: 'flex-1', - }, - 'x-component': 'Select', - 'x-reactions': { - dependencies: [ - '.accessNodeType', - '.accessNodeProcessId#dataSource', - '.accessNodeProcessId', - ], - fulfill: { - state: { - visible: - "{{'MANUALLY_SPECIFIED_BY_THE_USER_AGENT_GROUP' === $deps[0]}}", - }, - run: ` - let children = [] - - if ($deps[1] && $deps[2]) { - const item = $deps[1].find(item => item.accessNodeType === $deps[0] && item.value === $deps[2]) - children = (item && item.children) || [] - } - - $self.dataSource = [ - { - label:'${t( - 'packages_business_connection_form_automatic', - )}', - value: '' - } - ].concat(children) - - if ($self.value && !children.find(item => item.value === $self.value)) { - $self.value = null - } - `, - }, - }, - }, + 'x-component': 'TaskAgentSettings', + 'x-component-props': { + getOptions: formScope.$getTaskAgentOptions, + getConnectionCount: + formScope.$getTaskAgentConnectionCount, + refreshConnectionConfigs: + formScope.$refreshTaskAgentConfigs, }, }, + accessNodeType: { + type: 'string', + default: 'AUTOMATIC_PLATFORM_ALLOCATION', + 'x-display': 'hidden', + }, + accessNodeProcessId: { + type: 'string', + 'x-display': 'hidden', + }, + priorityProcessId: { + type: 'string', + default: '', + 'x-display': 'hidden', + }, }, }, }, @@ -1944,7 +1767,7 @@ defineExpose({ > .el-tabs__content { height: calc(100% - $tabHeight); - padding: 0 16px; + padding: 0 16px 24px; overflow: auto; .el-tab-pane { // height: 100%; diff --git a/packages/dag/src/components/form/agent-assignment-display/index.tsx b/packages/dag/src/components/form/agent-assignment-display/index.tsx new file mode 100644 index 0000000000..3d86fcbd8c --- /dev/null +++ b/packages/dag/src/components/form/agent-assignment-display/index.tsx @@ -0,0 +1,158 @@ +import { observer } from '@formily/reactive-vue' +import { useForm } from '@tap/form' +import i18n from '@tap/i18n' +import { defineComponent, inject } from 'vue' +import { getAccessNodeDisplayName } from '../../../utils/accessNodePolicy' +import './style.scss' + +const GROUP = 'MANUALLY_SPECIFIED_BY_THE_USER_AGENT_GROUP' +const ID_PATTERN = + /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i + +function displayAgentName(agent: any) { + const candidates = [ + agent?.agentName, + agent?.hostName, + agent?.systemInfo?.hostname, + agent?.hostname, + agent?.accessNodeName, + getAccessNodeDisplayName(agent), + ] + return candidates.find((name) => name && !ID_PATTERN.test(name)) +} + +export const AgentAssignmentDisplay = observer( + defineComponent({ + name: 'AgentAssignmentDisplay', + setup() { + const form = useForm() + const scope = inject('formScope') + + return () => { + const attrs = form.value.values.attrs || {} + const agentMap = scope?.$agentMap || {} + const agentOptions = scope?.$agents || [] + const processId = attrs.accessNodeProcessId + const processIds: string[] = attrs.accessNodeProcessIdList || [] + const selectedInfo = processId && agentMap[processId] + const isGroup = + attrs.accessNodeType === GROUP || + selectedInfo?.accessNodeType === GROUP || + Boolean(selectedInfo?.accessNodes) || + (!processId && processIds.length > 0) + const isAutomatic = !isGroup && !processId + + const groupOptions = agentOptions.filter( + (option: any) => option.accessNodeType === GROUP, + ) + const ids = new Set(processIds.filter(Boolean)) + const exactGroup = groupOptions.find( + (option: any) => processId && option.value === processId, + ) + const matchingGroups = groupOptions.filter( + (option: any) => + ids.size > 0 && + [...ids].every((id) => + option.children?.some((child: any) => child.value === id), + ), + ) + const group = isGroup + ? exactGroup || + groupOptions.find( + (option: any) => + ids.size > 0 && + option.children?.length === ids.size && + option.children.every((child: any) => ids.has(child.value)), + ) || + (matchingGroups.length === 1 ? matchingGroups[0] : undefined) + : undefined + const groupInfo = group && agentMap[group.value] + const members = + groupInfo?.accessNodes || + group?.children + ?.map((child: any) => agentMap[child.value]) + .filter(Boolean) || + [] + const manualAgent = selectedInfo + const title = isAutomatic + ? i18n.t('packages_dag_connection_form_automatic') + : isGroup + ? groupInfo?.accessNodeName || + group?.label?.replace(/（.*$/, '') || + i18n.t('packages_dag_agent_display_unavailable') + : displayAgentName(manualAgent) || + i18n.t('packages_dag_agent_display_unavailable') + const mode = isAutomatic + ? i18n.t('packages_dag_connection_form_automatic') + : isGroup + ? i18n.t('packages_business_connection_form_group') + : i18n.t('packages_dag_connection_form_manual') + + return ( +
+
+ + + {mode} + + {title} + + +
+ {isGroup && members.length > 0 && ( +
+ {members.map((member: any) => { + const name = + displayAgentName(agentMap[member.processId] || member) || + i18n.t('packages_dag_agent_display_unavailable') + const preferred = + attrs.priorityProcessId && + member.processId === attrs.priorityProcessId + return ( + + + {name} + {preferred && ( + + + + + {i18n.t('packages_dag_agent_display_priority')} + + )} + + ) + })} +
+ )} +
+ ) + } + }, + }), +) diff --git a/packages/dag/src/components/form/agent-assignment-display/style.scss b/packages/dag/src/components/form/agent-assignment-display/style.scss new file mode 100644 index 0000000000..add6ad8757 --- /dev/null +++ b/packages/dag/src/components/form/agent-assignment-display/style.scss @@ -0,0 +1,102 @@ +.agent-assignment { + padding: 12px 14px; + border-radius: 10px; + background: var(--bg-light); + + &__main { + display: flex; + align-items: center; + min-width: 0; + gap: 10px; + } + + &__icon { + display: inline-flex; + flex: none; + align-items: center; + justify-content: center; + width: 34px; + height: 34px; + border-radius: 9px; + color: var(--el-color-primary); + background: var(--el-color-primary-light-9); + } + + &__text { + display: flex; + flex-direction: column; + min-width: 0; + gap: 2px; + } + + &__mode { + color: var(--el-text-color-secondary); + font-size: 11px; + line-height: 16px; + } + + &__name { + overflow: hidden; + color: var(--el-text-color-primary); + font-size: 13px; + font-weight: 600; + line-height: 20px; + text-overflow: ellipsis; + white-space: nowrap; + } + + &__members { + display: flex; + flex-wrap: wrap; + gap: 6px; + margin-top: 12px; + padding-top: 12px; + border-top: 1px solid var(--el-border-color-lighter); + } + + &__member { + display: inline-flex; + align-items: center; + min-width: 0; + max-width: 100%; + gap: 6px; + padding: 4px 8px; + border: 1px solid var(--el-border-color-lighter); + border-radius: 8px; + background: var(--el-bg-color); + color: var(--el-text-color-regular); + font-size: 12px; + line-height: 18px; + } + + &__member-dot { + flex: none; + width: 5px; + height: 5px; + border-radius: 50%; + background: var(--el-color-success); + } + + &__member-dot--offline { + background: var(--el-text-color-placeholder); + } + + &__member-name { + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; + } + + &__member--preferred { + border-color: var(--el-color-primary-light-7); + } + + &__priority { + display: inline-flex; + align-items: center; + flex: none; + gap: 3px; + color: var(--el-color-primary); + font-size: 11px; + } +} diff --git a/packages/dag/src/components/form/index.ts b/packages/dag/src/components/form/index.ts index d7eef08b45..dfedf0ef57 100644 --- a/packages/dag/src/components/form/index.ts +++ b/packages/dag/src/components/form/index.ts @@ -37,3 +37,5 @@ export * from './data-preview' export * from './compare-schema' export * from './merge-table-cache' export * from './alarm-receiver' +export * from './agent-assignment-display' +export * from './task-agent-settings' diff --git a/packages/dag/src/components/form/task-agent-settings/index.tsx b/packages/dag/src/components/form/task-agent-settings/index.tsx new file mode 100644 index 0000000000..fbab4279bc --- /dev/null +++ b/packages/dag/src/components/form/task-agent-settings/index.tsx @@ -0,0 +1,242 @@ +import { observer } from '@formily/reactive-vue' +import { useForm } from '@tap/form' +import i18n from '@tap/i18n' +import { defineComponent, type PropType } from 'vue' +import './style.scss' + +const AUTOMATIC = 'AUTOMATIC_PLATFORM_ALLOCATION' +const MANUAL = 'MANUALLY_SPECIFIED_BY_THE_USER' +const GROUP = 'MANUALLY_SPECIFIED_BY_THE_USER_AGENT_GROUP' + +export const TaskAgentSettings = observer( + defineComponent({ + name: 'TaskAgentSettings', + props: { + getOptions: { + type: Function as PropType<(accessNodeType: string) => any[]>, + required: true, + }, + getConnectionCount: { + type: Function as PropType<() => number>, + required: true, + }, + refreshConnectionConfigs: { + type: Function as PropType<() => Promise>, + required: true, + }, + }, + setup(props) { + const formRef = useForm() + + const setMode = (mode: string) => { + const form = formRef.value + form.setValuesIn('accessNodeType', mode) + form.setValuesIn('accessNodeProcessId', '') + form.setValuesIn('priorityProcessId', '') + } + + const setProcessId = (processId: string) => { + const form = formRef.value + form.setValuesIn('accessNodeProcessId', processId) + form.setValuesIn('priorityProcessId', '') + } + + return () => { + const form = formRef.value + const values = form.values || {} + const mode = values.accessNodeType || AUTOMATIC + const isAutomatic = mode === AUTOMATIC + const isGroup = mode === GROUP + const options = props.getOptions(mode) || [] + const processId = values.accessNodeProcessId || '' + const selected = options.find( + (option: any) => option.value === processId, + ) + const priorityOptions = selected?.children || [] + const selectionInvalid = Boolean(processId && !selected) + const disabled = form.disabled + const connectionCount = props.getConnectionCount() || 0 + + return ( +
+
+ +
+
+ {isAutomatic + ? i18n.t('packages_dag_connection_form_automatic') + : i18n.t('packages_dag_task_agent_custom')} +
+
+ {isAutomatic + ? i18n.t('packages_dag_task_agent_auto_desc') + : connectionCount + ? i18n.t('packages_dag_task_agent_compatible_summary', { + connections: connectionCount, + agents: options.length, + }) + : i18n.t('packages_dag_task_agent_options_summary', { + agents: options.length, + })} +
+
+ {isAutomatic ? ( + setMode(MANUAL)} + > + {{ + icon: () => , + default: () => i18n.t('packages_dag_task_agent_customize'), + }} + + ) : ( + setMode(AUTOMATIC)} + > + {{ + icon: () => , + default: () => + i18n.t('packages_dag_task_agent_use_automatic'), + }} + + )} +
+ + {!isAutomatic && ( +
+ + + {options.length ? ( +
+ + + {isGroup && processId && !selectionInvalid && ( + + )} +
+ ) : ( +
+ + + +
+
+ {i18n.t('packages_dag_task_agent_no_compatible')} +
+
+ {i18n.t( + connectionCount + ? 'packages_dag_task_agent_no_compatible_desc' + : 'packages_dag_task_agent_no_available_desc', + )} +
+
+
+ )} + + {selectionInvalid && ( + + )} +
+ )} +
+ ) + } + }, + }), +) diff --git a/packages/dag/src/components/form/task-agent-settings/style.scss b/packages/dag/src/components/form/task-agent-settings/style.scss new file mode 100644 index 0000000000..40a8648bac --- /dev/null +++ b/packages/dag/src/components/form/task-agent-settings/style.scss @@ -0,0 +1,135 @@ +.task-agent-settings { + overflow: hidden; + border-radius: 10px; + background: var(--bg-light); + + &__header { + display: flex; + align-items: center; + min-height: 66px; + gap: 10px; + padding: 12px 14px; + } + + &__icon { + display: inline-flex; + flex: none; + align-items: center; + justify-content: center; + width: 34px; + height: 34px; + border-radius: 9px; + color: var(--el-color-primary); + background: var(--el-color-primary-light-9); + } + + &__summary { + flex: 1; + min-width: 0; + } + + &__title { + color: var(--el-text-color-primary); + font-size: 13px; + font-weight: 600; + line-height: 20px; + } + + &__description { + margin-top: 2px; + color: var(--el-text-color-secondary); + font-size: 12px; + line-height: 18px; + } + + &__body { + padding: 0 14px 14px; + } + + &__mode { + width: 100%; + margin-bottom: 14px; + + .el-segmented__item { + flex: 1; + } + } + + &__fields { + display: grid; + grid-template-columns: repeat(2, minmax(0, 1fr)); + gap: 12px; + } + + &__field { + display: block; + min-width: 0; + } + + &__label { + display: block; + margin-bottom: 6px; + color: var(--el-text-color-regular); + font-size: 12px; + font-weight: 500; + line-height: 18px; + } + + &__empty { + display: flex; + align-items: flex-start; + gap: 8px; + padding: 10px 12px; + border-radius: 8px; + color: var(--el-color-warning-dark-2); + background: var(--el-color-warning-light-9); + } + + &__empty-title { + font-size: 12px; + font-weight: 600; + line-height: 18px; + } + + &__empty-description { + margin-top: 2px; + color: var(--el-text-color-secondary); + font-size: 12px; + line-height: 18px; + } + + &__error { + display: flex; + align-items: center; + gap: 6px; + margin-top: 8px; + color: var(--el-color-danger); + font-size: 12px; + line-height: 18px; + } + + .el-button { + flex: none; + } + + .el-select.is-error .el-select__wrapper { + box-shadow: 0 0 0 1px var(--el-color-danger) inset; + } +} + +@media (max-width: 640px) { + .task-agent-settings { + &__header { + align-items: flex-start; + flex-wrap: wrap; + } + + &__header > .el-button { + margin-left: 44px; + } + + &__fields { + grid-template-columns: 1fr; + } + } +} diff --git a/packages/dag/src/components/materialized-view/Node.vue b/packages/dag/src/components/materialized-view/Node.vue index 7bfcbab569..1b6f76b0a0 100644 --- a/packages/dag/src/components/materialized-view/Node.vue +++ b/packages/dag/src/components/materialized-view/Node.vue @@ -585,6 +585,7 @@ function onConnectionSelect(connection) { connectionName: connection.name, connectionType: connection.connection_type, accessNodeProcessId: connection.accessNodeProcessId, + accessNodeProcessIdList: connection.accessNodeProcessIdList || [], pdkType: connection.pdkType, pdkHash: connection.pdkHash, capabilities: connection.capabilities || [], diff --git a/packages/dag/src/components/migration/LeftSider.vue b/packages/dag/src/components/migration/LeftSider.vue index 9e75e89ab7..9b957b9929 100644 --- a/packages/dag/src/components/migration/LeftSider.vue +++ b/packages/dag/src/components/migration/LeftSider.vue @@ -305,6 +305,7 @@ export default { connectionName: item.name, connectionType: item.connection_type, accessNodeProcessId: item.accessNodeProcessId, + accessNodeProcessIdList: item.accessNodeProcessIdList || [], priorityProcessId: item.priorityProcessId, pdkType: item.pdkType, pdkHash: item.pdkHash, diff --git a/packages/dag/src/components/steps/AdvancedSettingsStep.vue b/packages/dag/src/components/steps/AdvancedSettingsStep.vue index 666abcafa2..819d39c65f 100644 --- a/packages/dag/src/components/steps/AdvancedSettingsStep.vue +++ b/packages/dag/src/components/steps/AdvancedSettingsStep.vue @@ -17,9 +17,22 @@ import i18n from '@tap/i18n' import axios from 'axios' import { ElMessageBox, ElMessage as Message } from 'element-plus' import { debounce } from 'lodash-es' -import { computed, defineComponent, inject, onBeforeUnmount, ref } from 'vue' +import { + computed, + defineComponent, + inject, + onBeforeUnmount, + ref, + watch, +} from 'vue' import { useRouter } from 'vue-router' import { useStore } from 'vuex' +import { useConnectionAgentConfigs } from '../../composables/useConnectionAgentConfigs' +import { + createAccessNodeInfoMap, + getAccessNodeDisplayName, + getCompatibleAccessNodeOptions, +} from '../../utils/accessNodePolicy' import SchemaForm from '../SchemaForm.vue' function getSystemAlarmWarningSchema(index, alarmKey) { @@ -176,12 +189,15 @@ export default defineComponent({ return } - const map = { - [taskRef.value.dag.nodes[0].connectionId]: - taskRef.value.dag.nodes[0].attrs.connectionName, - [taskRef.value.dag.nodes[3].connectionId]: - taskRef.value.dag.nodes[3].attrs.connectionName, - } + const connectionNodeIds = new Set( + accessNodeProcessIdMap.value[field.value], + ) + const map = taskConnectionNodes.value + .filter((node) => connectionNodeIds.has(node.id)) + .reduce((result, node) => { + result[node.connectionId] = node.attrs.connectionName + return result + }, {}) const values = Object.values(map) @@ -206,6 +222,7 @@ export default defineComponent({ }, 500) const form = ref(null) + const selectedTaskAccessNodeType = ref(taskRef.value.accessNodeType) const starting = ref(false) const upgradeFeeVisible = ref(false) const upgradeFeeVisibleTips = ref(false) @@ -218,7 +235,7 @@ export default defineComponent({ const data = await findAccessNodeInfo() const mapNode = (item) => ({ value: item.processId, - label: `${item.agentName || item.hostName}（${ + label: `${getAccessNodeDisplayName(item) || '-'}（${ item.status === 'running' ? i18n.t('public_status_running') : i18n.t('public_agent_status_offline') @@ -232,7 +249,7 @@ export default defineComponent({ ) { return { value: item.processId, - label: `${item.accessNodeName}（${i18n.t('public_status_running')}：${ + label: `${getAccessNodeDisplayName(item) || '-'}（${i18n.t('public_status_running')}：${ item.accessNodes?.filter((ii) => ii.status === 'running') .length || 0 }）`, @@ -242,49 +259,18 @@ export default defineComponent({ } return mapNode(item) }) - scope.$agentMap = data.reduce( - (obj, item) => ((obj[item.processId] = item), obj), - {}, - ) + scope.$agentMap = createAccessNodeInfoMap(data) } const initTaskAgent = () => { - const size = accessNodeProcessIdArr.value.length - if (size >= 1) { - let currentId = taskRef.value.accessNodeProcessId - currentId = - currentId && accessNodeProcessIdArr.value.includes(currentId) - ? currentId - : accessNodeProcessIdArr.value[0] - taskRef.value.accessNodeType = - scope.$agentMap[currentId]?.accessNodeType || - 'MANUALLY_SPECIFIED_BY_THE_USER' - taskRef.value.accessNodeProcessId = currentId - - if ( - taskRef.value.accessNodeType === - 'MANUALLY_SPECIFIED_BY_THE_USER_AGENT_GROUP' - ) { - const nodeIds = accessNodeProcessIdMap.value[currentId] - let priorityProcessId = null - - nodeIds.some((id) => { - const node = taskRef.value.dag.nodes.find((node) => node.id === id) - if (node && node.attrs.priorityProcessId) { - priorityProcessId = node.attrs.priorityProcessId - return true - } - }) - - taskRef.value.priorityProcessId = priorityProcessId - } - } - + // Keep the task-level Agent policy independent from connection node + // attributes. The backend validates the selected Agent against the + // connection-policy intersection when saving. form.value.setFieldState('*(accessNodeType,accessNodeProcessId)', { - disabled: size === 1, + disabled: false, }) - form.value.setFieldState('accessNodeProcessId', { + form.value.setFieldState('*(accessNodeProcessId)', { dataSource: accessNodeProcessList.value, }) } @@ -301,36 +287,83 @@ export default defineComponent({ }) }, 500) + const taskConnectionNodes = computed(() => + (taskRef.value.dag?.nodes || []).filter((node) => + ['database', 'table'].includes(node.type), + ), + ) + + const connectionIds = computed(() => + [ + ...new Set(taskConnectionNodes.value.map((node) => node.connectionId)), + ].filter(Boolean), + ) + const { connectionAgentConfigs } = useConnectionAgentConfigs(connectionIds) + + const connectionAgentConfigList = computed(() => + taskConnectionNodes.value.map((node) => ({ + ...(node.attrs || {}), + ...(connectionAgentConfigs.value[node.connectionId] || {}), + })), + ) + const accessNodeProcessIdMap = computed(() => { - return [taskRef.value.dag.nodes[0], taskRef.value.dag.nodes[3]].reduce( - (map, node) => { - const { accessNodeProcessId } = node.attrs - if (accessNodeProcessId) { - let nodeIdArr = map[accessNodeProcessId] - - if (!nodeIdArr) { - nodeIdArr = map[accessNodeProcessId] = [] - } + return taskConnectionNodes.value.reduce((map, node, index) => { + const config = connectionAgentConfigList.value[index] + const processIds = config.accessNodeProcessId + ? [config.accessNodeProcessId] + : config.accessNodeProcessIdList || [] + + processIds.forEach((processId) => { + const nodeIds = (map[processId] ||= []) + nodeIds.push(node.id) + }) - nodeIdArr.push(node.id) - } - return map - }, - {}, - ) + return map + }, {}) }) - const accessNodeProcessIdArr = computed(() => { - return Object.keys(accessNodeProcessIdMap.value) - }) + const accessNodeProcessList = computed(() => + getCompatibleAccessNodeOptions( + scope.$agents || [], + connectionAgentConfigList.value, + selectedTaskAccessNodeType.value, + ), + ) + + watch( + accessNodeProcessList, + (dataSource = []) => { + form.value?.setFieldState('*(accessNodeProcessId)', { dataSource }) + }, + { deep: true, immediate: true }, + ) + + const getTaskSaveErrorMessage = (error) => { + const responseData = error?.response?.data ?? error + if (typeof responseData === 'string') return responseData + const validationData = + responseData?.code === 'Task.ListWarnMessage' + ? responseData.data + : (responseData?.data ?? responseData) + const validationItems = Array.isArray(validationData) + ? validationData + : Object.values(validationData || {}).flatMap((value) => + Array.isArray(value) ? value : [value], + ) + const validationMessage = validationItems.find( + (item) => item?.msg || item?.message, + ) - const accessNodeProcessList = computed(() => { - const agents = scope.$agents.filter( - (item) => item.accessNodeType === taskRef.value.accessNodeType, + return ( + validationMessage?.msg || + validationMessage?.message || + responseData?.message || + responseData?.msg || + error?.message || + error?.msg ) - if (!accessNodeProcessIdArr.value.length) return agents - return agents.filter((item) => !!accessNodeProcessIdMap.value[item.value]) - }) + } const schema = { type: 'object', @@ -1980,24 +2013,39 @@ export default defineComponent({ } const onTaskChange = debounce(async () => { - const data = await patchTask( - { - ...taskRef.value, - // id: taskRef.value.id, - // editVersion: taskRef.value.editVersion, - pageVersion: pageVersionRef.value, - // dag: taskRef.value.dag - }, - { - silenceMessage: true, - }, - ) - - // 防止触发 FormValuesChange - // const rawObj = raw(taskRef.value) - taskRef.value.editVersion = data.editVersion + try { + const data = await patchTask( + { + ...taskRef.value, + // Formily may keep the latest settings in form.values without + // mutating the injected task object. Include both so the task-level + // Agent policy reaches TM's save validator. + ...(form.value?.values || {}), + // id: taskRef.value.id, + // editVersion: taskRef.value.editVersion, + pageVersion: pageVersionRef.value, + // dag: taskRef.value.dag + }, + { + silenceMessage: true, + }, + ) - console.log('onTaskChange') + // 防止触发 FormValuesChange + // const rawObj = raw(taskRef.value) + taskRef.value.editVersion = data.editVersion + } catch (error) { + // Agent policy conflicts are returned by TM during the debounced task + // save. The request intentionally silences the global toast, so surface + // the validation error here only when this is not already part of the + // explicit start/save flow. The explicit flow will show one message. + if (!starting.value) { + showErrorMessage({ + msg: getTaskSaveErrorMessage(error), + stack: error?.stack, + }) + } + } }, 100) let dispose @@ -2067,6 +2115,9 @@ export default defineComponent({ // }) form.value.addEffects('watchForm', () => { + onFieldValueChange('accessNodeType', (field) => { + selectedTaskAccessNodeType.value = field.value + }) // onFormValuesChange(form => { // // onTaskChange() // console.log('onFormValuesChange', form.values) @@ -2124,7 +2175,7 @@ export default defineComponent({ }), )}`, ) - .then(async (data) => { + .then((data) => { const { items = [] } = data if (items.some((t) => t.status === 'Stopped')) { @@ -2147,26 +2198,10 @@ export default defineComponent({ error, msg = i18n.t('packages_dag_src_editor_chucuole'), ) => { - const code = error?.code + const code = error?.response?.data?.code ?? error?.code if (code === 'Task.ListWarnMessage') { - const names = [] - if (error.data) { - const keys = Object.keys(error.data) - keys.forEach((key) => { - const node = taskRef.value.dag.nodes.find(({ id }) => id === key) - if (node) { - names.push(node.name) - } - }) - if (!names.length && keys.length && msg) { - // 兼容错误信息id不是节点id的情况 - const msg = error.data[keys[0]][0]?.msg - if (msg) { - Message.error(msg) - return - } - } - } + Message.error(getTaskSaveErrorMessage(error) || msg) + return } else if (code === 'Task.OldVersion') { ElMessageBox.confirm( '', @@ -2185,7 +2220,10 @@ export default defineComponent({ ) { handleShowUpgradeDialog(error) } else { - showErrorMessage(error) + showErrorMessage({ + msg: getTaskSaveErrorMessage(error), + stack: error?.stack, + }) } } @@ -2194,9 +2232,15 @@ export default defineComponent({ let isOk = false try { - await saveAndStartTask(taskRef.value, { - silenceMessage: true, - }) + await saveAndStartTask( + { + ...taskRef.value, + ...(form.value?.values || {}), + }, + { + silenceMessage: true, + }, + ) isOk = true store.dispatch('setGuideComplete') diff --git a/packages/dag/src/components/steps/SourceStep.vue b/packages/dag/src/components/steps/SourceStep.vue index d4d4d4d788..391711f5b5 100644 --- a/packages/dag/src/components/steps/SourceStep.vue +++ b/packages/dag/src/components/steps/SourceStep.vue @@ -269,6 +269,7 @@ export default defineComponent({ connectionName: data.name, connectionType: data.connection_type, accessNodeProcessId: data.accessNodeProcessId, + accessNodeProcessIdList: data.accessNodeProcessIdList || [], priorityProcessId: data.priorityProcessId, pdkType: data.pdkType, pdkHash: data.pdkHash, diff --git a/packages/dag/src/composables/useCanvasOperation.ts b/packages/dag/src/composables/useCanvasOperation.ts index a80c0c892e..d8abfb465e 100644 --- a/packages/dag/src/composables/useCanvasOperation.ts +++ b/packages/dag/src/composables/useCanvasOperation.ts @@ -999,50 +999,9 @@ export function useCanvasOperation() { // --- Agent 校验 --- const validateAgent = () => { - let someErrorMsg: string | undefined - const nodes = allNodes.value.filter( - (node: any) => node.type === 'database' || node.type === 'table', - ) - const accessNodeProcessIdArr = [ - ...nodes.reduce((set: Set, item: any) => { - item.attrs.accessNodeProcessId && - set.add(item.attrs.accessNodeProcessId) - return set - }, new Set()), - ] - - if (accessNodeProcessIdArr.length > 1) { - const chooseId = dataflowStore.dataflow.accessNodeProcessId - if (!chooseId) { - someErrorMsg = t('packages_dag_mixins_editor_suoshuage') - } else { - let isError = false - const agent = (formScope as any)?.$agentMap?.[chooseId] - nodes.forEach((node: any) => { - if ( - node.attrs.accessNodeProcessId && - chooseId !== node.attrs.accessNodeProcessId - ) { - setNodeErrorMsg({ - id: node.id, - msg: t('packages_dag_mixins_editor_gaijiedianbuzhi', { - val1: agent?.hostName, - val2: agent?.ip, - }), - }) - isError = true - } - }) - if (isError) someErrorMsg = t('packages_dag_mixins_editor_suoshuage') - } - } else if (accessNodeProcessIdArr.length === 1) { - const agentId = accessNodeProcessIdArr[0] as string - dataflowStore.dataflow.accessNodeType = - (formScope as any)?.$agentMap?.[agentId]?.accessNodeType || - 'MANUALLY_SPECIFIED_BY_THE_USER' - dataflowStore.dataflow.accessNodeProcessId = agentId - } - return someErrorMsg + // Connection Agent policies are validated by TM on save. Do not infer or + // write the task-level Agent from DAG node attributes here. + return undefined } // --- validateSetting (需要外部传入 configPanel ref) --- @@ -1672,7 +1631,18 @@ export function useCanvasOperation() { batchStartTasks([dataflowStore.dataflow.id], { silenceMessage: true, }) - .then(() => { + .then((response) => { + const payload = response?.data ?? response + const failure = Array.isArray(payload) + ? payload.find((item) => item?.code && item.code !== 'ok') + : payload?.code && payload.code !== 'ok' + ? payload + : null + if (failure) { + buried(buriedCode, { result: false }) + handleError(failure) + return + } buried(buriedCode, { result: true }) router.push({ name: monitorRoute.value, diff --git a/packages/dag/src/composables/useConnectionAgentConfigs.ts b/packages/dag/src/composables/useConnectionAgentConfigs.ts new file mode 100644 index 0000000000..816244b941 --- /dev/null +++ b/packages/dag/src/composables/useConnectionAgentConfigs.ts @@ -0,0 +1,93 @@ +import { fetchConnections } from '@tap/api/src/core/connections' +import { onBeforeUnmount, onMounted, ref, watch, type Ref } from 'vue' + +type ConnectionAgentConfig = { + accessNodeType?: string + accessNodeProcessId?: string + accessNodeProcessIdList?: string[] + trueAccessNodeProcessIdList?: string[] +} + +export function useConnectionAgentConfigs(connectionIds: Ref) { + const connectionAgentConfigs = ref>({}) + let requestSequence = 0 + + const refreshConnectionAgentConfigs = async () => { + const ids = [...new Set(connectionIds.value.filter(Boolean))] + const requestId = ++requestSequence + + if (!ids.length) { + connectionAgentConfigs.value = {} + return + } + + try { + const { items = [] } = await fetchConnections({ + where: { id: { $in: ids } }, + fields: { + id: 1, + accessNodeType: 1, + accessNodeProcessId: 1, + accessNodeProcessIdList: 1, + trueAccessNodeProcessIdList: 1, + }, + size: ids.length, + }) + + if (requestId !== requestSequence) return + + connectionAgentConfigs.value = items.reduce( + (configs: Record, connection: any) => { + const id = connection.id || connection._id + if (id) { + const config: ConnectionAgentConfig = {} + if (connection.accessNodeType !== undefined) { + config.accessNodeType = connection.accessNodeType + } + if (connection.accessNodeProcessId !== undefined) { + config.accessNodeProcessId = connection.accessNodeProcessId + } + if ( + connection.trueAccessNodeProcessIdList !== undefined || + connection.accessNodeProcessIdList !== undefined + ) { + config.accessNodeProcessIdList = + connection.trueAccessNodeProcessIdList || + connection.accessNodeProcessIdList || + [] + } + configs[id] = config + } + return configs + }, + {}, + ) + } catch { + // Keep the last successful snapshot; task save validation remains authoritative. + } + } + + const refreshWhenVisible = async () => { + if (document.visibilityState === 'visible') { + await refreshConnectionAgentConfigs() + } + } + + watch(connectionIds, refreshConnectionAgentConfigs, { + deep: true, + immediate: true, + }) + + onMounted(() => { + window.addEventListener('focus', refreshWhenVisible) + document.addEventListener('visibilitychange', refreshWhenVisible) + }) + + onBeforeUnmount(() => { + window.removeEventListener('focus', refreshWhenVisible) + document.removeEventListener('visibilitychange', refreshWhenVisible) + requestSequence += 1 + }) + + return { connectionAgentConfigs, refreshConnectionAgentConfigs } +} diff --git a/packages/dag/src/composables/useDnD.ts b/packages/dag/src/composables/useDnD.ts index 027b1bbee5..270639aad8 100644 --- a/packages/dag/src/composables/useDnD.ts +++ b/packages/dag/src/composables/useDnD.ts @@ -13,6 +13,7 @@ export const makeNode = ( connectionName: connection.name, connectionType: connection.connection_type, accessNodeProcessId: connection.accessNodeProcessId, + accessNodeProcessIdList: connection.accessNodeProcessIdList || [], priorityProcessId: connection.priorityProcessId, pdkType: connection.pdkType, pdkHash: connection.pdkHash, diff --git a/packages/dag/src/composables/useFormScope.ts b/packages/dag/src/composables/useFormScope.ts index 47204f0fee..fd62ad0ff3 100644 --- a/packages/dag/src/composables/useFormScope.ts +++ b/packages/dag/src/composables/useFormScope.ts @@ -28,6 +28,10 @@ import { computed, reactive, watch } from 'vue' import { useRoute } from 'vue-router' import { useStore } from 'vuex' import { useDataflowStore } from '../stores/dataflow.store' +import { + createAccessNodeInfoMap, + getAccessNodeDisplayName, +} from '../utils/accessNodePolicy' import type { Field } from '@formily/core' function addDeclaredCompleter(editor, tools, params1) { @@ -140,7 +144,7 @@ export function useFormScope({ canvasRef }) { const data = await findAccessNodeInfo() const mapNode = (item) => ({ value: item.processId, - label: `${item.agentName || item.hostName}（${ + label: `${getAccessNodeDisplayName(item) || '-'}（${ item.status === 'running' ? t('public_status_running') : t('public_agent_status_offline') @@ -154,7 +158,7 @@ export function useFormScope({ canvasRef }) { ) { return { value: item.processId, - label: `${item.accessNodeName}（${t('public_status_running')}：${ + label: `${getAccessNodeDisplayName(item) || '-'}（${t('public_status_running')}：${ item.accessNodes?.filter((ii) => ii.status === 'running').length || 0 }）`, @@ -164,10 +168,7 @@ export function useFormScope({ canvasRef }) { } return mapNode(item) }) - scope.$agentMap = data.reduce( - (obj, item) => ((obj[item.processId] = item), obj), - {}, - ) + scope.$agentMap = createAccessNodeInfoMap(data) } const findNodeById = dataflowStore.findNodeById @@ -467,6 +468,7 @@ export function useFormScope({ canvasRef }) { databaseType: item.database_type, connectionType: item.connection_type, accessNodeProcessId: item.accessNodeProcessId, + accessNodeProcessIdList: item.accessNodeProcessIdList || [], } }) @@ -579,12 +581,20 @@ export function useFormScope({ canvasRef }) { }) const connectionType = form.getValuesIn('attrs.connectionType') const accessNodeProcessId = form.getValuesIn('attrs.accessNodeProcessId') + const accessNodeProcessIdList = form.getValuesIn( + 'attrs.accessNodeProcessIdList', + ) const connectionName = form.getValuesIn('attrs.connectionName') connectionType !== item.connectionType && form.setValuesIn('attrs.connectionType', item.connectionType) accessNodeProcessId !== item.accessNodeProcessId && form.setValuesIn('attrs.accessNodeProcessId', item.accessNodeProcessId) + !isEqual(accessNodeProcessIdList, item.accessNodeProcessIdList) && + form.setValuesIn( + 'attrs.accessNodeProcessIdList', + item.accessNodeProcessIdList || [], + ) connectionName !== item.name && form.setValuesIn('attrs.connectionName', item.name) }, @@ -886,6 +896,10 @@ export function useFormScope({ canvasRef }) { 'attrs.accessNodeProcessId', connection.accessNodeProcessId, ) + form.setValuesIn( + 'attrs.accessNodeProcessIdList', + connection.accessNodeProcessIdList || [], + ) form.setValuesIn('attrs.accessNodeType', connection.accessNodeType) form.setValuesIn('attrs.priorityProcessId', connection.priorityProcessId) form.setValuesIn('attrs.connectionName', connection.name) diff --git a/packages/dag/src/locale/lang/en.js b/packages/dag/src/locale/lang/en.js index 2ca17fad68..7278b49e77 100644 --- a/packages/dag/src/locale/lang/en.js +++ b/packages/dag/src/locale/lang/en.js @@ -67,6 +67,23 @@ export default { packages_dag_agent_setting_from: 'Agent settings derive from the following', packages_dag_connection_form_automatic: 'Platform automatic allocation', packages_dag_connection_form_manual: 'User specified manually', + packages_dag_agent_display_unavailable: 'Agent information unavailable', + packages_dag_agent_display_priority: 'Preferred', + packages_dag_task_agent_custom: 'Custom Agent scheduling', + packages_dag_task_agent_auto_desc: + 'The platform selects an available Agent based on the task connections', + packages_dag_task_agent_customize: 'Customize', + packages_dag_task_agent_use_automatic: 'Use automatic allocation', + packages_dag_task_agent_compatible_summary: + 'Filtered by {connections} connections, {agents} options available', + packages_dag_task_agent_options_summary: '{agents} options available', + packages_dag_task_agent_no_compatible: 'No compatible Agent', + packages_dag_task_agent_no_compatible_desc: + 'The connection settings conflict. Update them or use automatic allocation.', + packages_dag_task_agent_no_available_desc: + 'No Agent is available for this selection. Check the Agent list or use automatic allocation.', + packages_dag_task_agent_selection_invalid: + 'The current selection no longer matches the connection settings. Select another option or use automatic allocation.', packages_dag_task_setting_automatic_ddl: 'Automatic DDL', packages_dag_task_setting_distinct_write_type: 're-rewriting mechanism', packages_dag_task_setting_transformer_concurrency: diff --git a/packages/dag/src/locale/lang/zh-CN.js b/packages/dag/src/locale/lang/zh-CN.js index 6cabf723e7..56ffd9cf56 100644 --- a/packages/dag/src/locale/lang/zh-CN.js +++ b/packages/dag/src/locale/lang/zh-CN.js @@ -59,6 +59,23 @@ export default { packages_dag_agent_setting_from: 'agent设置来源于以下连接', packages_dag_connection_form_automatic: '平台自动分配', packages_dag_connection_form_manual: '用户手动指定', + packages_dag_agent_display_unavailable: '暂无法获取 Agent 信息', + packages_dag_agent_display_priority: '优先', + packages_dag_task_agent_custom: '自定义 Agent 调度', + packages_dag_task_agent_auto_desc: + '平台将根据任务涉及的连接自动选择可用 Agent', + packages_dag_task_agent_customize: '自定义', + packages_dag_task_agent_use_automatic: '恢复自动分配', + packages_dag_task_agent_compatible_summary: + '已根据 {connections} 个连接筛选，可选择 {agents} 项', + packages_dag_task_agent_options_summary: '可选择 {agents} 项', + packages_dag_task_agent_no_compatible: '没有兼容的 Agent', + packages_dag_task_agent_no_compatible_desc: + '当前连接配置存在冲突，请调整连接设置或恢复平台自动分配。', + packages_dag_task_agent_no_available_desc: + '当前没有可选择的 Agent，请检查 Agent 列表或恢复平台自动分配。', + packages_dag_task_agent_selection_invalid: + '当前选择已不符合连接配置，请重新选择或恢复平台自动分配。', packages_dag_task_setting_automatic_ddl: '自动DDL', packages_dag_task_setting_distinct_write_type: '去重写入机制', packages_dag_task_setting_transformer_concurrency: '目标写入线程数', diff --git a/packages/dag/src/locale/lang/zh-TW.js b/packages/dag/src/locale/lang/zh-TW.js index d98f5ab064..ede4e84a6f 100644 --- a/packages/dag/src/locale/lang/zh-TW.js +++ b/packages/dag/src/locale/lang/zh-TW.js @@ -59,6 +59,23 @@ export default { packages_dag_agent_setting_from: '代理设置來自以下連結', packages_dag_connection_form_automatic: '平台自動分配', packages_dag_connection_form_manual: '用戶手動指定', + packages_dag_agent_display_unavailable: '暫無法取得 Agent 資訊', + packages_dag_agent_display_priority: '優先', + packages_dag_task_agent_custom: '自訂 Agent 調度', + packages_dag_task_agent_auto_desc: + '平台將根據任務涉及的連線自動選擇可用 Agent', + packages_dag_task_agent_customize: '自訂', + packages_dag_task_agent_use_automatic: '恢復自動分配', + packages_dag_task_agent_compatible_summary: + '已根據 {connections} 個連線篩選，可選擇 {agents} 項', + packages_dag_task_agent_options_summary: '可選擇 {agents} 項', + packages_dag_task_agent_no_compatible: '沒有相容的 Agent', + packages_dag_task_agent_no_compatible_desc: + '目前連線設定存在衝突，請調整連線設定或恢復平台自動分配。', + packages_dag_task_agent_no_available_desc: + '目前沒有可選擇的 Agent，請檢查 Agent 清單或恢復平台自動分配。', + packages_dag_task_agent_selection_invalid: + '目前選擇已不符合連線設定，請重新選擇或恢復平台自動分配。', packages_dag_task_setting_automatic_ddl: '自動DDL', packages_dag_task_setting_distinct_write_type: '去重寫入機制', packages_dag_task_setting_transformer_concurrency: '目標寫入線程數', diff --git a/packages/dag/src/mixins/editor.js b/packages/dag/src/mixins/editor.js index 5e0672751f..e40f8e597e 100644 --- a/packages/dag/src/mixins/editor.js +++ b/packages/dag/src/mixins/editor.js @@ -1496,55 +1496,9 @@ export default { * @returns {*} */ validateAgent() { - let someErrorMsg - const nodes = this.allNodes.filter( - (node) => node.type === 'database' || node.type === 'table', - ) - const accessNodeProcessIdArr = [ - ...nodes.reduce((set, item) => { - item.attrs.accessNodeProcessId && - set.add(item.attrs.accessNodeProcessId) - return set - }, new Set()), - ] - - if (accessNodeProcessIdArr.length > 1) { - // 所属agent节点冲突 - const chooseId = this.dataflow.accessNodeProcessId - - if (!chooseId) { - // someErrorMsg = `请配置任务运行agent` - someErrorMsg = i18n.t('packages_dag_mixins_editor_suoshuage') // 一样提示冲突 - } else { - let isError = false - const agent = this.scope.$agentMap[chooseId] - nodes.forEach((node) => { - if ( - node.attrs.accessNodeProcessId && - chooseId !== node.attrs.accessNodeProcessId - ) { - this.setNodeErrorMsg({ - id: node.id, - msg: i18n.t('packages_dag_mixins_editor_gaijiedianbuzhi', { - val1: agent.hostName, - val2: agent.ip, - }), - }) - isError = true - } - }) - isError && - (someErrorMsg = i18n.t('packages_dag_mixins_editor_suoshuage')) - } - } else if (accessNodeProcessIdArr.length === 1) { - // 如果画布上仅有一个所属agent，自动设置为任务的agent - const agentId = accessNodeProcessIdArr[0] - this.dataflow.accessNodeType = - this.scope.$agentMap[agentId]?.accessNodeType || - 'MANUALLY_SPECIFIED_BY_THE_USER' - this.dataflow.accessNodeProcessId = agentId - } - return someErrorMsg + // Connection Agent policies are validated by TM on save. Do not infer or + // write the task-level Agent from DAG node attributes here. + return undefined }, async validateSetting() { @@ -1593,7 +1547,10 @@ export default { const arr = [] if (node.$outputs.length) { node.$outputs.forEach((id) => { - console.log('this.loadLeafNode(this.nodeById(id))', this.loadLeafNode(this.nodeById(id))) // eslint-disable-line + console.log( + 'this.loadLeafNode(this.nodeById(id))', + this.loadLeafNode(this.nodeById(id)), + ) // eslint-disable-line arr.push(...this.loadLeafNode(this.nodeById(id))) }) } else { @@ -1674,10 +1631,7 @@ export default { let hasEnableDDL let hasEnableDDLAndIncreasesql let inBlacklist = false - const blacklist = [ - 'union_processor', - 'migrate_union_processor', - ] + const blacklist = ['union_processor', 'migrate_union_processor'] this.allNodes.forEach((node) => { // 开启了DDL if (node.ddlConfiguration === 'SYNCHRONIZATION') { @@ -2930,7 +2884,18 @@ export default { batchStartTasks([this.dataflow.id], { silenceMessage: true, }) - .then(() => { + .then((response) => { + const payload = response?.data ?? response + const failure = Array.isArray(payload) + ? payload.find((item) => item?.code && item.code !== 'ok') + : payload?.code && payload.code !== 'ok' + ? payload + : null + if (failure) { + this.buried(buriedCode, { result: false }) + this.handleError(failure) + return + } this.buried(buriedCode, { result: true }) this.gotoViewer() }) diff --git a/packages/dag/src/mixins/formScope.js b/packages/dag/src/mixins/formScope.js index 848de9e59d..24d7bbbf72 100644 --- a/packages/dag/src/mixins/formScope.js +++ b/packages/dag/src/mixins/formScope.js @@ -25,6 +25,10 @@ import { isPlainObj } from '@tap/shared' import axios from 'axios' import { isEmpty, isEqual, merge } from 'lodash-es' import { mapGetters, mapState } from 'vuex' +import { + createAccessNodeInfoMap, + getAccessNodeDisplayName, +} from '../utils/accessNodePolicy' const editorKeyboard = { handleKeyboard({ editor }, hash, keyString, keyCode, event) { @@ -353,6 +357,7 @@ export default { databaseType: item.database_type, connectionType: item.connection_type, accessNodeProcessId: item.accessNodeProcessId, + accessNodeProcessIdList: item.accessNodeProcessIdList || [], } }) @@ -466,6 +471,9 @@ export default { const accessNodeProcessId = form.getValuesIn( 'attrs.accessNodeProcessId', ) + const accessNodeProcessIdList = form.getValuesIn( + 'attrs.accessNodeProcessIdList', + ) const connectionName = form.getValuesIn('attrs.connectionName') connectionType !== item.connectionType && @@ -475,6 +483,11 @@ export default { 'attrs.accessNodeProcessId', item.accessNodeProcessId, ) + !isEqual(accessNodeProcessIdList, item.accessNodeProcessIdList) && + form.setValuesIn( + 'attrs.accessNodeProcessIdList', + item.accessNodeProcessIdList || [], + ) connectionName !== item.name && form.setValuesIn('attrs.connectionName', item.name) }, @@ -882,6 +895,8 @@ export default { const connectionType = form.getValuesIn('attrs.connectionType') || '' const accessNodeProcessId = form.getValuesIn('attrs.accessNodeProcessId') || '' + const accessNodeProcessIdList = + form.getValuesIn('attrs.accessNodeProcessIdList') || [] const accessNodeType = form.getValuesIn('attrs.accessNodeType') || '' const priorityProcessId = form.getValuesIn('attrs.priorityProcessId') || '' @@ -903,6 +918,14 @@ export default { 'attrs.accessNodeProcessId', connection.accessNodeProcessId, ) + !isEqual( + accessNodeProcessIdList, + connection.accessNodeProcessIdList, + ) && + form.setValuesIn( + 'attrs.accessNodeProcessIdList', + connection.accessNodeProcessIdList || [], + ) accessNodeType !== connection.accessNodeType && form.setValuesIn('attrs.accessNodeType', connection.accessNodeType) priorityProcessId !== connection.priorityProcessId && @@ -1309,7 +1332,7 @@ export default { const data = await findAccessNodeInfo() const mapNode = (item) => ({ value: item.processId, - label: `${item.agentName || item.hostName}（${ + label: `${getAccessNodeDisplayName(item) || '-'}（${ item.status === 'running' ? i18n.t('public_status_running') : i18n.t('public_agent_status_offline') @@ -1323,7 +1346,7 @@ export default { ) { return { value: item.processId, - label: `${item.accessNodeName}（${i18n.t('public_status_running')}：${ + label: `${getAccessNodeDisplayName(item) || '-'}（${i18n.t('public_status_running')}：${ item.accessNodes?.filter((ii) => ii.status === 'running') .length || 0 }）`, @@ -1333,10 +1356,7 @@ export default { } return mapNode(item) }) - this.scope.$agentMap = data.reduce( - (obj, item) => ((obj[item.processId] = item), obj), - {}, - ) + this.scope.$agentMap = createAccessNodeInfoMap(data) }, /** diff --git a/packages/dag/src/nodes/Database.js b/packages/dag/src/nodes/Database.js index 1b4b914d91..1e1bd9f91d 100644 --- a/packages/dag/src/nodes/Database.js +++ b/packages/dag/src/nodes/Database.js @@ -129,14 +129,16 @@ export class Database extends NodeType { 'x-decorator-props': { feedbackLayout: 'none', }, - 'x-component': 'PreviewText.Input', - 'x-component-props': { - content: `{{$agentMap[$self.value] ? $values.attrs.accessNodeType === 'MANUALLY_SPECIFIED_BY_THE_USER_AGENT_GROUP' ? \`\${$agentMap[$self.value].accessNodeName} (\${$agentMap[$self.value].processId})\` : \`\${$agentMap[$self.value].hostName}（\${$agentMap[$self.value].ip}）\` : "-"}}`, - }, + 'x-component': 'AgentAssignmentDisplay', 'x-reactions': { + dependencies: [ + 'attrs.accessNodeType', + 'attrs.accessNodeProcessIdList', + ], fulfill: { state: { - display: '{{!$self.value ? "hidden":"visible"}}', + display: + '{{($values.attrs.accessNodeType !== "AUTOMATIC_PLATFORM_ALLOCATION" && ($self.value || $deps[1]?.length)) ? "visible" : "hidden"}}', }, }, }, diff --git a/packages/dag/src/nodes/Table.js b/packages/dag/src/nodes/Table.js index ec43b093b2..2110d10db2 100644 --- a/packages/dag/src/nodes/Table.js +++ b/packages/dag/src/nodes/Table.js @@ -311,14 +311,16 @@ export class Table extends NodeType { type: 'string', title: i18n.t('packages_dag_nodes_database_suoshuage'), 'x-decorator': 'FormItem', - 'x-component': 'PreviewText.Input', - 'x-component-props': { - content: `{{$agentMap[$self.value] ? $values.attrs.accessNodeType === 'MANUALLY_SPECIFIED_BY_THE_USER_AGENT_GROUP' ? \`\${$agentMap[$self.value].accessNodeName} (\${$agentMap[$self.value].processId})\` : \`\${$agentMap[$self.value].hostName}（\${$agentMap[$self.value].ip}）\` : "-"}}`, - }, + 'x-component': 'AgentAssignmentDisplay', 'x-reactions': { + dependencies: [ + 'attrs.accessNodeType', + 'attrs.accessNodeProcessIdList', + ], fulfill: { state: { - display: '{{!$self.value ? "hidden":"visible"}}', + display: + '{{($values.attrs.accessNodeType !== "AUTOMATIC_PLATFORM_ALLOCATION" && ($self.value || $deps[1]?.length)) ? "visible" : "hidden"}}', }, }, }, diff --git a/packages/dag/src/utils/accessNodePolicy.ts b/packages/dag/src/utils/accessNodePolicy.ts new file mode 100644 index 0000000000..0085729677 --- /dev/null +++ b/packages/dag/src/utils/accessNodePolicy.ts @@ -0,0 +1,138 @@ +type AccessNodeOption = { + value: string + accessNodeType: string + disabled?: boolean + children?: AccessNodeOption[] +} + +type AccessNodeInfo = { + processId?: string + agentName?: string + hostName?: string + hostname?: string + displayName?: string + systemInfo?: { + hostname?: string + } + accessNodeName?: string + accessNodes?: AccessNodeInfo[] +} + +type ConnectionAgentConfig = { + accessNodeType?: string + accessNodeProcessId?: string + accessNodeProcessIdList?: string[] +} + +const AUTOMATIC_ALLOCATION = 'AUTOMATIC_PLATFORM_ALLOCATION' +const AGENT_GROUP_KEY_PREFIX = '__agent_group__:' + +function getAgentGroupKey(processIds: string[] = []) { + return `${AGENT_GROUP_KEY_PREFIX}${[...new Set(processIds.filter(Boolean))].sort().join(',')}` +} + +export function getAccessNodeDisplayName(accessNode: AccessNodeInfo = {}) { + return ( + accessNode.agentName || + accessNode.displayName || + accessNode.accessNodeName || + accessNode.hostName || + accessNode.systemInfo?.hostname || + accessNode.hostname + ) +} + +export function createAccessNodeInfoMap(accessNodeInfo: AccessNodeInfo[] = []) { + return accessNodeInfo.reduce>((map, item) => { + if (item.processId) { + map[item.processId] = { + ...item, + displayName: getAccessNodeDisplayName(item), + } + } + item.accessNodes?.forEach((accessNode) => { + if (accessNode.processId) { + map[accessNode.processId] = { + ...accessNode, + displayName: getAccessNodeDisplayName(accessNode), + } + } + }) + if (item.accessNodes?.length) { + map[ + getAgentGroupKey( + item.accessNodes.map(({ processId }) => processId || ''), + ) + ] = { ...item, displayName: getAccessNodeDisplayName(item) } + } + return map + }, {}) +} + +export function getCompatibleAccessNodeOptions( + agents: AccessNodeOption[], + connections: ConnectionAgentConfig[], + accessNodeType: string, +) { + const getConfiguredProcessIds = (connection: ConnectionAgentConfig) => + connection.accessNodeProcessId + ? [connection.accessNodeProcessId] + : connection.accessNodeProcessIdList?.filter(Boolean) || [] + + const candidates = agents.filter( + (agent) => agent.accessNodeType === accessNodeType, + ) + const configuredConnections = connections.filter( + (connection) => + getConfiguredProcessIds(connection).length > 0 && + connection.accessNodeType !== AUTOMATIC_ALLOCATION, + ) + + if (!configuredConnections.length) return candidates + + const allowedAgentSets = configuredConnections.map((connection) => { + return new Set( + getConfiguredProcessIds(connection).flatMap((processId) => { + const configuredAgent = agents.find( + (agent) => agent.value === processId, + ) + + return configuredAgent?.children + ? configuredAgent.children + .filter((child) => !child.disabled) + .map((child) => child.value) + : [processId] + }), + ) + }) + + const compatibleAgentIds = allowedAgentSets + .slice(1) + .reduce( + (intersection, allowedIds) => + new Set([...intersection].filter((id) => allowedIds.has(id))), + allowedAgentSets[0]!, + ) + + return candidates.flatMap((candidate) => { + if (candidate.children) { + const availableChildren = candidate.children.filter( + (child) => !child.disabled, + ) + const children = candidate.children.filter((child) => + compatibleAgentIds.has(child.value), + ) + + if ( + !availableChildren.length || + availableChildren.some((child) => !compatibleAgentIds.has(child.value)) + ) { + return [] + } + + return children.length ? [{ ...candidate, children }] : [] + } + + return compatibleAgentIds.has(candidate.value) ? [candidate] : [] + }) +}