| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.seatunnel.engine.server; |
| |
| import org.apache.seatunnel.api.common.metrics.JobMetrics; |
| import org.apache.seatunnel.api.common.metrics.RawJobMetrics; |
| import org.apache.seatunnel.api.event.EventHandler; |
| import org.apache.seatunnel.api.event.EventProcessor; |
| import org.apache.seatunnel.common.utils.ExceptionUtils; |
| import org.apache.seatunnel.common.utils.SeaTunnelException; |
| import org.apache.seatunnel.common.utils.StringFormatUtils; |
| import org.apache.seatunnel.engine.common.Constant; |
| import org.apache.seatunnel.engine.common.config.EngineConfig; |
| import org.apache.seatunnel.engine.common.config.server.ConnectorJarStorageConfig; |
| import org.apache.seatunnel.engine.common.exception.JobException; |
| import org.apache.seatunnel.engine.common.exception.JobNotFoundException; |
| import org.apache.seatunnel.engine.common.exception.SavePointFailedException; |
| import org.apache.seatunnel.engine.common.exception.SeaTunnelEngineException; |
| import org.apache.seatunnel.engine.common.utils.PassiveCompletableFuture; |
| import org.apache.seatunnel.engine.core.job.JobDAGInfo; |
| import org.apache.seatunnel.engine.core.job.JobInfo; |
| import org.apache.seatunnel.engine.core.job.JobResult; |
| import org.apache.seatunnel.engine.core.job.JobStatus; |
| import org.apache.seatunnel.engine.core.job.PipelineStatus; |
| import org.apache.seatunnel.engine.server.dag.physical.PhysicalVertex; |
| import org.apache.seatunnel.engine.server.dag.physical.PipelineLocation; |
| import org.apache.seatunnel.engine.server.dag.physical.SubPlan; |
| import org.apache.seatunnel.engine.server.event.JobEventHttpReportHandler; |
| import org.apache.seatunnel.engine.server.event.JobEventProcessor; |
| import org.apache.seatunnel.engine.server.execution.ExecutionState; |
| import org.apache.seatunnel.engine.server.execution.TaskExecutionState; |
| import org.apache.seatunnel.engine.server.execution.TaskGroupLocation; |
| import org.apache.seatunnel.engine.server.execution.TaskLocation; |
| import org.apache.seatunnel.engine.server.master.JobHistoryService; |
| import org.apache.seatunnel.engine.server.master.JobMaster; |
| import org.apache.seatunnel.engine.server.metrics.JobMetricsUtil; |
| import org.apache.seatunnel.engine.server.metrics.SeaTunnelMetricsContext; |
| import org.apache.seatunnel.engine.server.resourcemanager.ResourceManager; |
| import org.apache.seatunnel.engine.server.resourcemanager.ResourceManagerFactory; |
| import org.apache.seatunnel.engine.server.resourcemanager.resource.SlotProfile; |
| import org.apache.seatunnel.engine.server.service.jar.ConnectorPackageService; |
| import org.apache.seatunnel.engine.server.task.operation.GetMetricsOperation; |
| import org.apache.seatunnel.engine.server.utils.NodeEngineUtil; |
| |
| import com.google.common.util.concurrent.ThreadFactoryBuilder; |
| import com.hazelcast.cluster.Address; |
| import com.hazelcast.config.Config; |
| import com.hazelcast.core.HazelcastInstanceNotActiveException; |
| import com.hazelcast.internal.serialization.Data; |
| import com.hazelcast.internal.services.MembershipServiceEvent; |
| import com.hazelcast.logging.ILogger; |
| import com.hazelcast.map.IMap; |
| import com.hazelcast.ringbuffer.Ringbuffer; |
| import com.hazelcast.spi.impl.NodeEngineImpl; |
| import lombok.NonNull; |
| |
| import java.util.ArrayList; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.concurrent.CompletableFuture; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.ScheduledExecutorService; |
| import java.util.concurrent.ThreadPoolExecutor; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicLong; |
| import java.util.stream.Collectors; |
| |
| import static org.apache.seatunnel.engine.server.metrics.JobMetricsUtil.toJobMetricsMap; |
| |
| public class CoordinatorService { |
| private final NodeEngineImpl nodeEngine; |
| private final ILogger logger; |
| |
| private volatile ResourceManager resourceManager; |
| |
| private JobHistoryService jobHistoryService; |
| |
| /** |
| * IMap key is jobId and value is {@link JobInfo}. Tuple2 key is JobMaster init timestamp and |
| * value is the jobImmutableInformation which is sent by client when submit job |
| * |
| * <p>This IMap is used to recovery runningJobInfoIMap in JobMaster when a new master node |
| * active |
| */ |
| private IMap<Long, JobInfo> runningJobInfoIMap; |
| |
| /** |
| * IMap key is one of jobId {@link |
| * org.apache.seatunnel.engine.server.dag.physical.PipelineLocation} and {@link |
| * org.apache.seatunnel.engine.server.execution.TaskGroupLocation} |
| * |
| * <p>The value of IMap is one of {@link JobStatus} {@link PipelineStatus} {@link |
| * org.apache.seatunnel.engine.server.execution.ExecutionState} |
| * |
| * <p>This IMap is used to recovery runningJobStateIMap in JobMaster when a new master node |
| * active |
| */ |
| IMap<Object, Object> runningJobStateIMap; |
| |
| /** |
| * IMap key is one of jobId {@link |
| * org.apache.seatunnel.engine.server.dag.physical.PipelineLocation} and {@link |
| * org.apache.seatunnel.engine.server.execution.TaskGroupLocation} |
| * |
| * <p>The value of IMap is one of {@link |
| * org.apache.seatunnel.engine.server.dag.physical.PhysicalPlan} stateTimestamps {@link |
| * org.apache.seatunnel.engine.server.dag.physical.SubPlan} stateTimestamps {@link |
| * org.apache.seatunnel.engine.server.dag.physical.PhysicalVertex} stateTimestamps |
| * |
| * <p>This IMap is used to recovery runningJobStateTimestampsIMap in JobMaster when a new master |
| * node active |
| */ |
| IMap<Object, Long[]> runningJobStateTimestampsIMap; |
| |
| /** |
| * key: job id; <br> |
| * value: job master; |
| */ |
| private Map<Long, JobMaster> runningJobMasterMap = new ConcurrentHashMap<>(); |
| |
| /** |
| * IMap key is {@link PipelineLocation} |
| * |
| * <p>The value of IMap is map of {@link TaskGroupLocation} and the {@link SlotProfile} it used. |
| * |
| * <p>This IMap is used to recovery ownedSlotProfilesIMap in JobMaster when a new master node |
| * active |
| */ |
| private IMap<PipelineLocation, Map<TaskGroupLocation, SlotProfile>> ownedSlotProfilesIMap; |
| |
| private IMap<Long, HashMap<TaskLocation, SeaTunnelMetricsContext>> metricsImap; |
| |
| /** If this node is a master node */ |
| private volatile boolean isActive = false; |
| |
| private ExecutorService executorService; |
| |
| private final SeaTunnelServer seaTunnelServer; |
| |
| private final ScheduledExecutorService masterActiveListener; |
| |
| private final EngineConfig engineConfig; |
| |
| private ConnectorPackageService connectorPackageService; |
| |
| private EventProcessor eventProcessor; |
| |
| public CoordinatorService( |
| @NonNull NodeEngineImpl nodeEngine, |
| @NonNull SeaTunnelServer seaTunnelServer, |
| EngineConfig engineConfig) { |
| this.nodeEngine = nodeEngine; |
| this.logger = nodeEngine.getLogger(getClass()); |
| this.executorService = |
| Executors.newCachedThreadPool( |
| new ThreadFactoryBuilder() |
| .setNameFormat("seatunnel-coordinator-service-%d") |
| .build()); |
| this.seaTunnelServer = seaTunnelServer; |
| this.engineConfig = engineConfig; |
| masterActiveListener = Executors.newSingleThreadScheduledExecutor(); |
| masterActiveListener.scheduleAtFixedRate( |
| this::checkNewActiveMaster, 0, 100, TimeUnit.MILLISECONDS); |
| } |
| |
| private JobEventProcessor createJobEventProcessor( |
| String reportHttpEndpoint, |
| Map<String, String> reportHttpHeaders, |
| NodeEngineImpl nodeEngine) { |
| List<EventHandler> handlers = |
| EventProcessor.loadEventHandlers(Thread.currentThread().getContextClassLoader()); |
| |
| if (reportHttpEndpoint != null) { |
| String ringBufferName = "zeta-job-event"; |
| int maxBufferCapacity = 2000; |
| nodeEngine |
| .getHazelcastInstance() |
| .getConfig() |
| .addRingBufferConfig( |
| new Config() |
| .getRingbufferConfig(ringBufferName) |
| .setCapacity(maxBufferCapacity) |
| .setBackupCount(0) |
| .setAsyncBackupCount(1) |
| .setTimeToLiveSeconds(0)); |
| Ringbuffer ringbuffer = nodeEngine.getHazelcastInstance().getRingbuffer(ringBufferName); |
| JobEventHttpReportHandler httpReportHandler = |
| new JobEventHttpReportHandler( |
| reportHttpEndpoint, reportHttpHeaders, ringbuffer); |
| handlers.add(httpReportHandler); |
| } |
| logger.info("Loaded event handlers: " + handlers); |
| JobEventProcessor eventProcessor = new JobEventProcessor(handlers); |
| return eventProcessor; |
| } |
| |
| public JobHistoryService getJobHistoryService() { |
| return jobHistoryService; |
| } |
| |
| public JobMaster getJobMaster(Long jobId) { |
| return runningJobMasterMap.get(jobId); |
| } |
| |
| public EventProcessor getEventProcessor() { |
| return eventProcessor; |
| } |
| |
| // On the new master node |
| // 1. If runningJobStateIMap.get(jobId) == null and runningJobInfoIMap.get(jobId) != null. We |
| // will do |
| // runningJobInfoIMap.remove(jobId) |
| // |
| // 2. If runningJobStateIMap.get(jobId) != null and the value equals JobStatus End State. We |
| // need new a |
| // JobMaster and generate PhysicalPlan again and then try to remove all of PipelineLocation |
| // and |
| // TaskGroupLocation key in the runningJobStateIMap. |
| // |
| // 3. If runningJobStateIMap.get(jobId) != null and the value equals JobStatus.SCHEDULED. We |
| // need cancel the job |
| // and then call submitJob(long jobId, Data jobImmutableInformation) to resubmit it. |
| // |
| // 4. If runningJobStateIMap.get(jobId) != null and the value is CANCELING or RUNNING. We need |
| // recover the JobMaster |
| // from runningJobStateIMap and then waiting for it complete. |
| private void initCoordinatorService() { |
| runningJobInfoIMap = |
| nodeEngine.getHazelcastInstance().getMap(Constant.IMAP_RUNNING_JOB_INFO); |
| runningJobStateIMap = |
| nodeEngine.getHazelcastInstance().getMap(Constant.IMAP_RUNNING_JOB_STATE); |
| runningJobStateTimestampsIMap = |
| nodeEngine.getHazelcastInstance().getMap(Constant.IMAP_STATE_TIMESTAMPS); |
| ownedSlotProfilesIMap = |
| nodeEngine.getHazelcastInstance().getMap(Constant.IMAP_OWNED_SLOT_PROFILES); |
| metricsImap = nodeEngine.getHazelcastInstance().getMap(Constant.IMAP_RUNNING_JOB_METRICS); |
| |
| jobHistoryService = |
| new JobHistoryService( |
| runningJobStateIMap, |
| logger, |
| runningJobMasterMap, |
| nodeEngine.getHazelcastInstance().getMap(Constant.IMAP_FINISHED_JOB_STATE), |
| nodeEngine |
| .getHazelcastInstance() |
| .getMap(Constant.IMAP_FINISHED_JOB_METRICS), |
| nodeEngine |
| .getHazelcastInstance() |
| .getMap(Constant.IMAP_FINISHED_JOB_VERTEX_INFO), |
| engineConfig.getHistoryJobExpireMinutes()); |
| eventProcessor = |
| createJobEventProcessor( |
| engineConfig.getEventReportHttpApi(), |
| engineConfig.getEventReportHttpHeaders(), |
| nodeEngine); |
| |
| // If the user has configured the connector package service, create it on the master node. |
| ConnectorJarStorageConfig connectorJarStorageConfig = |
| engineConfig.getConnectorJarStorageConfig(); |
| if (connectorJarStorageConfig.getEnable()) { |
| connectorPackageService = new ConnectorPackageService(seaTunnelServer); |
| } |
| |
| List<CompletableFuture<Void>> collect = |
| runningJobInfoIMap.entrySet().stream() |
| .map( |
| entry -> |
| CompletableFuture.runAsync( |
| () -> { |
| logger.info( |
| String.format( |
| "begin restore job (%s) from master active switch", |
| entry.getKey())); |
| try { |
| restoreJobFromMasterActiveSwitch( |
| entry.getKey(), entry.getValue()); |
| } catch (Exception e) { |
| logger.severe(e); |
| } |
| logger.info( |
| String.format( |
| "restore job (%s) from master active switch finished", |
| entry.getKey())); |
| }, |
| executorService)) |
| .collect(Collectors.toList()); |
| |
| try { |
| CompletableFuture<Void> voidCompletableFuture = |
| CompletableFuture.allOf(collect.toArray(new CompletableFuture[0])); |
| voidCompletableFuture.get(); |
| } catch (Exception e) { |
| throw new SeaTunnelEngineException(e); |
| } |
| } |
| |
| private void restoreJobFromMasterActiveSwitch(@NonNull Long jobId, @NonNull JobInfo jobInfo) { |
| if (runningJobStateIMap.get(jobId) == null) { |
| runningJobInfoIMap.remove(jobId); |
| return; |
| } |
| |
| JobStatus jobStatus = (JobStatus) runningJobStateIMap.get(jobId); |
| JobMaster jobMaster = |
| new JobMaster( |
| jobInfo.getJobImmutableInformation(), |
| nodeEngine, |
| executorService, |
| getResourceManager(), |
| getJobHistoryService(), |
| runningJobStateIMap, |
| runningJobStateTimestampsIMap, |
| ownedSlotProfilesIMap, |
| runningJobInfoIMap, |
| metricsImap, |
| engineConfig, |
| seaTunnelServer); |
| |
| try { |
| jobMaster.init(runningJobInfoIMap.get(jobId).getInitializationTimestamp(), true); |
| } catch (Exception e) { |
| throw new SeaTunnelEngineException(String.format("Job id %s init failed", jobId), e); |
| } |
| |
| String jobFullName = jobMaster.getPhysicalPlan().getJobFullName(); |
| runningJobMasterMap.put(jobId, jobMaster); |
| |
| logger.info( |
| String.format( |
| "The restore %s is in %s state, restore pipeline and take over this job running", |
| jobFullName, jobStatus)); |
| CompletableFuture.runAsync( |
| () -> { |
| try { |
| jobMaster |
| .getPhysicalPlan() |
| .getPipelineList() |
| .forEach(SubPlan::restorePipelineState); |
| jobMaster.run(); |
| } finally { |
| // voidCompletableFuture will be cancelled when zeta master node |
| // shutdown to simulate master failure, |
| // don't update runningJobMasterMap is this case. |
| if (!jobMaster.getJobMasterCompleteFuture().isCompletedExceptionally()) { |
| runningJobMasterMap.remove(jobId); |
| } |
| } |
| }, |
| executorService); |
| } |
| |
| private void checkNewActiveMaster() { |
| try { |
| if (!isActive && this.seaTunnelServer.isMasterNode()) { |
| logger.info( |
| "This node become a new active master node, begin init coordinator service"); |
| if (this.executorService.isShutdown()) { |
| this.executorService = |
| Executors.newCachedThreadPool( |
| new ThreadFactoryBuilder() |
| .setNameFormat("seatunnel-coordinator-service-%d") |
| .build()); |
| } |
| initCoordinatorService(); |
| isActive = true; |
| } else if (isActive && !this.seaTunnelServer.isMasterNode()) { |
| isActive = false; |
| logger.info( |
| "This node become leave active master node, begin clear coordinator service"); |
| clearCoordinatorService(); |
| } |
| } catch (Exception e) { |
| isActive = false; |
| logger.severe(ExceptionUtils.getMessage(e)); |
| throw new SeaTunnelEngineException("check new active master error, stop loop", e); |
| } |
| } |
| |
| public synchronized void clearCoordinatorService() { |
| // interrupt all JobMaster |
| runningJobMasterMap.values().forEach(JobMaster::interrupt); |
| executorService.shutdownNow(); |
| runningJobMasterMap.clear(); |
| |
| try { |
| executorService.awaitTermination(20, TimeUnit.SECONDS); |
| } catch (InterruptedException e) { |
| throw new SeaTunnelEngineException("wait clean executor service error", e); |
| } |
| |
| if (resourceManager != null) { |
| resourceManager.close(); |
| } |
| |
| try { |
| if (eventProcessor != null) { |
| eventProcessor.close(); |
| } |
| } catch (Exception e) { |
| throw new SeaTunnelEngineException("close event processor error", e); |
| } |
| } |
| |
| /** Lazy load for resource manager */ |
| public ResourceManager getResourceManager() { |
| if (resourceManager == null) { |
| synchronized (this) { |
| if (resourceManager == null) { |
| ResourceManager manager = |
| new ResourceManagerFactory(nodeEngine).getResourceManager(); |
| manager.init(); |
| resourceManager = manager; |
| } |
| } |
| } |
| return resourceManager; |
| } |
| |
| /** call by client to submit job */ |
| public PassiveCompletableFuture<Void> submitJob(long jobId, Data jobImmutableInformation) { |
| CompletableFuture<Void> jobSubmitFuture = new CompletableFuture<>(); |
| |
| // Check if the current jobID is already running. If so, complete the submission |
| // successfully. |
| // This avoids potential issues like redundant job restores or other anomalies. |
| if (getJobMaster(jobId) != null) { |
| logger.warning( |
| String.format( |
| "The job %s is currently running; no need to submit again.", jobId)); |
| jobSubmitFuture.complete(null); |
| return new PassiveCompletableFuture<>(jobSubmitFuture); |
| } |
| |
| JobMaster jobMaster = |
| new JobMaster( |
| jobImmutableInformation, |
| this.nodeEngine, |
| executorService, |
| getResourceManager(), |
| getJobHistoryService(), |
| runningJobStateIMap, |
| runningJobStateTimestampsIMap, |
| ownedSlotProfilesIMap, |
| runningJobInfoIMap, |
| metricsImap, |
| engineConfig, |
| seaTunnelServer); |
| executorService.submit( |
| () -> { |
| try { |
| runningJobInfoIMap.put( |
| jobId, |
| new JobInfo(System.currentTimeMillis(), jobImmutableInformation)); |
| runningJobMasterMap.put(jobId, jobMaster); |
| jobMaster.init( |
| runningJobInfoIMap.get(jobId).getInitializationTimestamp(), false); |
| // We specify that when init is complete, the submitJob is complete |
| jobSubmitFuture.complete(null); |
| } catch (Throwable e) { |
| logger.severe( |
| String.format( |
| "submit job %s error %s ", |
| jobId, ExceptionUtils.getMessage(e))); |
| jobSubmitFuture.completeExceptionally(e); |
| } |
| if (!jobSubmitFuture.isCompletedExceptionally()) { |
| try { |
| jobMaster.run(); |
| } finally { |
| // voidCompletableFuture will be cancelled when zeta master node |
| // shutdown to simulate master failure, |
| // don't update runningJobMasterMap is this case. |
| if (!jobMaster.getJobMasterCompleteFuture().isCancelled()) { |
| runningJobMasterMap.remove(jobId); |
| } |
| } |
| } else { |
| runningJobInfoIMap.remove(jobId); |
| runningJobMasterMap.remove(jobId); |
| } |
| }); |
| return new PassiveCompletableFuture<>(jobSubmitFuture); |
| } |
| |
| public PassiveCompletableFuture<Void> savePoint(long jobId) { |
| CompletableFuture<Void> voidCompletableFuture = new CompletableFuture<>(); |
| if (!runningJobMasterMap.containsKey(jobId)) { |
| SavePointFailedException exception = |
| new SavePointFailedException( |
| "The job with id '" + jobId + "' not running, save point failed"); |
| logger.warning(exception); |
| voidCompletableFuture.completeExceptionally(exception); |
| } else { |
| voidCompletableFuture = |
| new PassiveCompletableFuture<>( |
| CompletableFuture.supplyAsync( |
| () -> { |
| JobMaster runningJobMaster = runningJobMasterMap.get(jobId); |
| if (!runningJobMaster.savePoint().join()) { |
| throw new SavePointFailedException( |
| "The job with id '" |
| + jobId |
| + "' save point failed"); |
| } |
| return null; |
| }, |
| executorService)); |
| } |
| return new PassiveCompletableFuture<>(voidCompletableFuture); |
| } |
| |
| public PassiveCompletableFuture<JobResult> waitForJobComplete(long jobId) { |
| JobMaster runningJobMaster = runningJobMasterMap.get(jobId); |
| if (runningJobMaster == null) { |
| // Because operations on Imap cannot be performed within Operation. |
| CompletableFuture<JobHistoryService.JobState> jobStateFuture = |
| CompletableFuture.supplyAsync( |
| () -> { |
| return jobHistoryService.getJobDetailState(jobId); |
| }, |
| executorService); |
| JobHistoryService.JobState jobState = null; |
| try { |
| jobState = jobStateFuture.get(); |
| } catch (Exception e) { |
| throw new SeaTunnelEngineException("get job state error", e); |
| } |
| |
| CompletableFuture<JobResult> future = new CompletableFuture<>(); |
| if (jobState == null) future.complete(new JobResult(JobStatus.UNKNOWABLE, null)); |
| else |
| future.complete(new JobResult(jobState.getJobStatus(), jobState.getErrorMessage())); |
| return new PassiveCompletableFuture<>(future); |
| } else { |
| return new PassiveCompletableFuture<>(runningJobMaster.getJobMasterCompleteFuture()); |
| } |
| } |
| |
| public PassiveCompletableFuture<Void> cancelJob(long jodId) { |
| JobMaster runningJobMaster = runningJobMasterMap.get(jodId); |
| if (runningJobMaster == null) { |
| CompletableFuture<Void> future = new CompletableFuture<>(); |
| future.complete(null); |
| return new PassiveCompletableFuture<>(future); |
| } else { |
| return new PassiveCompletableFuture<>( |
| CompletableFuture.supplyAsync( |
| () -> { |
| runningJobMaster.cancelJob(); |
| return null; |
| }, |
| executorService)); |
| } |
| } |
| |
| public JobStatus getJobStatus(long jobId) { |
| JobMaster runningJobMaster = runningJobMasterMap.get(jobId); |
| if (runningJobMaster == null) { |
| JobHistoryService.JobState jobDetailState = jobHistoryService.getJobDetailState(jobId); |
| return null == jobDetailState ? JobStatus.UNKNOWABLE : jobDetailState.getJobStatus(); |
| } |
| JobStatus jobStatus = runningJobMaster.getJobStatus(); |
| if (jobStatus == null) { |
| return jobHistoryService.getFinishedJobStateImap().get(jobId).getJobStatus(); |
| } |
| return jobStatus; |
| } |
| |
| public JobMetrics getJobMetrics(long jobId) { |
| JobMaster runningJobMaster = runningJobMasterMap.get(jobId); |
| if (runningJobMaster == null) { |
| return jobHistoryService.getJobMetrics(jobId); |
| } |
| JobMetrics jobMetrics = JobMetricsUtil.toJobMetrics(runningJobMaster.getCurrJobMetrics()); |
| JobMetrics jobMetricsImap = jobHistoryService.getJobMetrics(jobId); |
| return jobMetricsImap != null ? jobMetricsImap.merge(jobMetrics) : jobMetrics; |
| } |
| |
| public Map<Long, JobMetrics> getRunningJobMetrics() { |
| final Set<Long> runningJobIds = runningJobMasterMap.keySet(); |
| |
| Set<Address> addresses = new HashSet<>(); |
| ownedSlotProfilesIMap.forEach( |
| (pipelineLocation, ownedSlotProfilesIMap) -> { |
| if (runningJobIds.contains(pipelineLocation.getJobId())) { |
| ownedSlotProfilesIMap |
| .values() |
| .forEach( |
| ownedSlotProfile -> { |
| addresses.add(ownedSlotProfile.getWorker()); |
| }); |
| } |
| }); |
| |
| List<RawJobMetrics> metrics = new ArrayList<>(); |
| |
| addresses.forEach( |
| address -> { |
| try { |
| if (nodeEngine.getClusterService().getMember(address) != null) { |
| RawJobMetrics rawJobMetrics = |
| (RawJobMetrics) |
| NodeEngineUtil.sendOperationToMemberNode( |
| nodeEngine, |
| new GetMetricsOperation(runningJobIds), |
| address) |
| .get(); |
| metrics.add(rawJobMetrics); |
| } |
| } |
| // HazelcastInstanceNotActiveException. It means that the node is |
| // offline, so waiting for the taskGroup to restore can be successful |
| catch (HazelcastInstanceNotActiveException e) { |
| logger.warning( |
| String.format( |
| "get metrics with exception: %s.", |
| ExceptionUtils.getMessage(e))); |
| } catch (Exception e) { |
| throw new SeaTunnelException(e.getMessage()); |
| } |
| }); |
| |
| Map<Long, JobMetrics> longJobMetricsMap = toJobMetricsMap(metrics); |
| |
| longJobMetricsMap.forEach( |
| (jobId, jobMetrics) -> { |
| JobMetrics jobMetricsImap = jobHistoryService.getJobMetrics(jobId); |
| if (jobMetricsImap != null) { |
| longJobMetricsMap.put(jobId, jobMetricsImap.merge(jobMetrics)); |
| } |
| }); |
| |
| return longJobMetricsMap; |
| } |
| |
| public JobDAGInfo getJobInfo(long jobId) { |
| JobDAGInfo jobInfo = jobHistoryService.getJobDAGInfo(jobId); |
| if (jobInfo != null) { |
| return jobInfo; |
| } |
| return runningJobMasterMap.get(jobId).getJobDAGInfo(); |
| } |
| |
| /** |
| * When TaskGroup ends, it is called by {@link TaskExecutionService} to notify JobMaster the |
| * TaskGroup's state. |
| */ |
| public void updateTaskExecutionState(TaskExecutionState taskExecutionState) { |
| logger.info( |
| String.format( |
| "Received task end from execution %s, state %s", |
| taskExecutionState.getTaskGroupLocation(), |
| taskExecutionState.getExecutionState())); |
| TaskGroupLocation taskGroupLocation = taskExecutionState.getTaskGroupLocation(); |
| JobMaster runningJobMaster = runningJobMasterMap.get(taskGroupLocation.getJobId()); |
| if (runningJobMaster == null) { |
| throw new JobNotFoundException( |
| String.format("Job %s not running", taskGroupLocation.getJobId())); |
| } |
| runningJobMaster.updateTaskExecutionState(taskExecutionState); |
| } |
| |
| public void shutdown() { |
| if (masterActiveListener != null) { |
| masterActiveListener.shutdownNow(); |
| } |
| clearCoordinatorService(); |
| } |
| |
| /** return true if this node is a master node and the coordinator service init finished. */ |
| public boolean isCoordinatorActive() { |
| return isActive; |
| } |
| |
| public void failedTaskOnMemberRemoved(MembershipServiceEvent event) { |
| Address lostAddress = event.getMember().getAddress(); |
| runningJobMasterMap.forEach( |
| (aLong, jobMaster) -> { |
| jobMaster |
| .getPhysicalPlan() |
| .getPipelineList() |
| .forEach( |
| subPlan -> { |
| makeTasksFailed( |
| subPlan.getCoordinatorVertexList(), lostAddress); |
| makeTasksFailed( |
| subPlan.getPhysicalVertexList(), lostAddress); |
| }); |
| }); |
| } |
| |
| private void makeTasksFailed( |
| @NonNull List<PhysicalVertex> physicalVertexList, @NonNull Address lostAddress) { |
| physicalVertexList.forEach( |
| physicalVertex -> { |
| Address deployAddress = physicalVertex.getCurrentExecutionAddress(); |
| ExecutionState executionState = physicalVertex.getExecutionState(); |
| if (null != deployAddress |
| && deployAddress.equals(lostAddress) |
| && (executionState.equals(ExecutionState.DEPLOYING) |
| || executionState.equals(ExecutionState.RUNNING) |
| || executionState.equals(ExecutionState.CANCELING))) { |
| TaskGroupLocation taskGroupLocation = physicalVertex.getTaskGroupLocation(); |
| physicalVertex.updateStateByExecutionService( |
| new TaskExecutionState( |
| taskGroupLocation, |
| ExecutionState.FAILED, |
| new JobException( |
| String.format( |
| "The taskGroup(%s) deployed node(%s) offline", |
| taskGroupLocation, lostAddress)))); |
| } |
| }); |
| } |
| |
| public void memberRemoved(MembershipServiceEvent event) { |
| if (isCoordinatorActive()) { |
| this.getResourceManager().memberRemoved(event); |
| } |
| this.failedTaskOnMemberRemoved(event); |
| } |
| |
| public void printExecutionInfo() { |
| ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) executorService; |
| int activeCount = threadPoolExecutor.getActiveCount(); |
| int corePoolSize = threadPoolExecutor.getCorePoolSize(); |
| int maximumPoolSize = threadPoolExecutor.getMaximumPoolSize(); |
| int poolSize = threadPoolExecutor.getPoolSize(); |
| long completedTaskCount = threadPoolExecutor.getCompletedTaskCount(); |
| long taskCount = threadPoolExecutor.getTaskCount(); |
| logger.info( |
| StringFormatUtils.formatTable( |
| "CoordinatorService Thread Pool Status", |
| "activeCount", |
| activeCount, |
| "corePoolSize", |
| corePoolSize, |
| "maximumPoolSize", |
| maximumPoolSize, |
| "poolSize", |
| poolSize, |
| "completedTaskCount", |
| completedTaskCount, |
| "taskCount", |
| taskCount)); |
| } |
| |
| public void printJobDetailInfo() { |
| AtomicLong createdJobCount = new AtomicLong(); |
| AtomicLong scheduledJobCount = new AtomicLong(); |
| AtomicLong runningJobCount = new AtomicLong(); |
| AtomicLong failingJobCount = new AtomicLong(); |
| AtomicLong failedJobCount = new AtomicLong(); |
| AtomicLong cancellingJobCount = new AtomicLong(); |
| AtomicLong canceledJobCount = new AtomicLong(); |
| AtomicLong finishedJobCount = new AtomicLong(); |
| |
| if (runningJobInfoIMap != null) { |
| runningJobInfoIMap |
| .keySet() |
| .forEach( |
| jobId -> { |
| if (runningJobStateIMap.get(jobId) != null) { |
| JobStatus jobStatus = |
| (JobStatus) runningJobStateIMap.get(jobId); |
| switch (jobStatus) { |
| case CREATED: |
| createdJobCount.addAndGet(1); |
| break; |
| case SCHEDULED: |
| scheduledJobCount.addAndGet(1); |
| break; |
| case RUNNING: |
| runningJobCount.addAndGet(1); |
| break; |
| case FAILING: |
| failingJobCount.addAndGet(1); |
| break; |
| case FAILED: |
| failedJobCount.addAndGet(1); |
| break; |
| case CANCELING: |
| cancellingJobCount.addAndGet(1); |
| break; |
| case CANCELED: |
| canceledJobCount.addAndGet(1); |
| break; |
| case FINISHED: |
| finishedJobCount.addAndGet(1); |
| break; |
| default: |
| } |
| } |
| }); |
| } |
| |
| logger.info( |
| StringFormatUtils.formatTable( |
| "Job info detail", |
| "createdJobCount", |
| createdJobCount, |
| "scheduledJobCount", |
| scheduledJobCount, |
| "runningJobCount", |
| runningJobCount, |
| "failingJobCount", |
| failingJobCount, |
| "failedJobCount", |
| failedJobCount, |
| "cancellingJobCount", |
| cancellingJobCount, |
| "canceledJobCount", |
| canceledJobCount, |
| "finishedJobCount", |
| finishedJobCount)); |
| } |
| |
| public ConnectorPackageService getConnectorPackageService() { |
| if (connectorPackageService == null) { |
| throw new SeaTunnelEngineException( |
| "The user is not configured to enable connector package service, can not get connector package service service from master node."); |
| } |
| return connectorPackageService; |
| } |
| } |
