Flink JobManager HA on Kubernetes

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123

public interface HighAvailabilityServices extends ClientHighAvailabilityServices {
  // 获取 resource manager 的 leader retriever
	LeaderRetrievalService getResourceManagerLeaderRetriever();

  // 获取 dispatcher 的 leader retriever
	LeaderRetrievalService getDispatcherLeaderRetriever();

	/**
	 * Gets the leader retriever for the job JobMaster which is responsible for the given job.
	 *
	 * @param jobID The identifier of the job.
	 * @param defaultJobManagerAddress JobManager address which will be returned by
	 *                              a static leader retrieval service.
	 * @return Leader retrieval service to retrieve the job manager for the given job
	 */
	LeaderRetrievalService getJobManagerLeaderRetriever(JobID jobID, String defaultJobManagerAddress);

	/**
	 * Gets the leader election service for the cluster's resource manager.
	 *
	 * @return Leader election service for the resource manager leader election
	 */
	LeaderElectionService getResourceManagerLeaderElectionService();

	/**
	 * Gets the leader election service for the cluster's dispatcher.
	 *
	 * @return Leader election service for the dispatcher leader election
	 */
	LeaderElectionService getDispatcherLeaderElectionService();

	/**
	 * Gets the leader election service for the given job.
	 *
	 * @param jobID The identifier of the job running the election.
	 * @return Leader election service for the job manager leader election
	 */
	LeaderElectionService getJobManagerLeaderElectionService(JobID jobID);

	/**
	 * Gets the checkpoint recovery factory for the job manager.
	 *
	 * @return Checkpoint recovery factory
	 */
	CheckpointRecoveryFactory getCheckpointRecoveryFactory();

	/**
	 * Gets the submitted job graph store for the job manager.
	 *
	 * @return Submitted job graph store
	 * @throws Exception if the submitted job graph store could not be created
	 */
	JobGraphStore getJobGraphStore() throws Exception;

	/**
	 * Gets the registry that holds information about whether jobs are currently running.
	 *
	 * @return Running job registry to retrieve running jobs
	 */
	RunningJobsRegistry getRunningJobsRegistry() throws Exception;

	/**
	 * Creates the BLOB store in which BLOBs are stored in a highly-available fashion.
	 *
	 * @return Blob store
	 * @throws IOException if the blob store could not be created
	 */
	BlobStore createBlobStore() throws IOException;

	/**
	 * Gets the leader election service for the cluster's rest endpoint.
	 *
	 * @return the leader election service used by the cluster's rest endpoint
	 */
	default LeaderElectionService getClusterRestEndpointLeaderElectionService() {
		// for backwards compatibility we delegate to getWebMonitorLeaderElectionService
		// all implementations of this interface should override getClusterRestEndpointLeaderElectionService, though
		return getWebMonitorLeaderElectionService();
	}

	
	default LeaderRetrievalService getClusterRestEndpointLeaderRetriever() {
		// for backwards compatibility we delegate to getWebMonitorLeaderRetriever
		// all implementations of this interface should override getClusterRestEndpointLeaderRetriever, though
		return getWebMonitorLeaderRetriever();
	}

	// ------------------------------------------------------------------------
	//  Shutdown and Cleanup
	// ------------------------------------------------------------------------

	/**
	 * Closes the high availability services, releasing all resources.
	 *
	 * <p>This method <b>does not delete or clean up</b> any data stored in external stores
	 * (file systems, ZooKeeper, etc). Another instance of the high availability
	 * services will be able to recover the job.
	 *
	 * <p>If an exception occurs during closing services, this method will attempt to
	 * continue closing other services and report exceptions only after all services
	 * have been attempted to be closed.
	 *
	 * @throws Exception Thrown, if an exception occurred while closing these services.
	 */
	
	void close() throws Exception;

	/**
	 * Closes the high availability services (releasing all resources) and deletes
	 * all data stored by these services in external stores.
	 *
	 * <p>After this method was called, the any job or session that was managed by
	 * these high availability services will be unrecoverable.
	 *
	 * <p>If an exception occurs during cleanup, this method will attempt to
	 * continue the cleanup and report exceptions only after all cleanup steps have
	 * been attempted.
	 *
	 * @throws Exception Thrown, if an exception occurred while closing these services
	 *                   or cleaning up data stored by them.
	 */
	void closeAndCleanupAllData() throws Exception;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

public interface LeaderElectionService {
  // 启动 leader 选举服务，内部在选举结束之后会根据成功或者失败调用 LeaderContender 的指定的回调函数
  // 除此之外还会将 contender 加入到 listener 中，以便在之后的选举流程中使用
	void start(LeaderContender contender) throws Exception;

  // 停止 leader 选举服务
	void stop() throws Exception;

  // 确认竞争者接受了 leader 委派并发布其 leader 地址
	void confirmLeadership(UUID leaderSessionID, String leaderAddress);

  // 返回在 leaderSessionId 内，与该 service 绑定的竞争者是否是 leader
	boolean hasLeadership(@Nonnull UUID leaderSessionId);
}

// 需要参与 leader 选举的服务都需要实现该接口
public interface LeaderContender {
  // 被选为 leader 时的回调函数
	void grantLeadership(UUID leaderSessionID);

  // 之前的 leader 被剥夺 leader 角色时的调用函数
	void revokeLeadership();

  // 错误处理
	void handleError(Exception exception);

	default String getDescription() {
		return "LeaderContender: " + getClass().getSimpleName();
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

public interface LeaderRetrievalService {
  // 启动 service，并且将 listener 加到回调队列中，当出现新的 leader 时进行回调。
	void start(LeaderRetrievalListener listener) throws Exception;

	void stop() throws Exception;
}

public interface LeaderRetrievalListener {
  // 当新的 leader 被选出来时，会通过回调该方法进行通知
	void notifyLeaderAddress(@Nullable String leaderAddress, @Nullable UUID leaderSessionID);

	void handleError(Exception exception);
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

public interface RunningJobsRegistry {
	enum JobSchedulingStatus {
		PENDING,
		RUNNING,
		DONE;
	}

  // 将作业状态设置为 Running
	void setJobRunning(JobID jobID) throws IOException;

  // 将作业状态设置为完成
	void setJobFinished(JobID jobID) throws IOException;

  // 获得作业的 schedule 状态
	JobSchedulingStatus getJobSchedulingStatus(JobID jobID) throws IOException;

	void clearJob(JobID jobID) throws IOException;
}

1
2
3

leaderLatch = new LeaderLatch(client, latchPath);	// 创建 LeaderLatch
leaderLatch.addListener(this);										// 添加回调对象
leaderLatch.start();				

1
2
3
4
5

public interface LeaderLatchListener {
    void isLeader();	// leader 回调

    void notLeader();	// 非 leader 回调
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

 /**
 * Gets the leader election service for the cluster's resource manager.
 */
LeaderElectionService getResourceManagerLeaderElectionService();
 /**
 * Gets the leader election service for the cluster's dispatcher.
 */
LeaderElectionService getDispatcherLeaderElectionService();
 /**
 * Gets the leader election service for the given job.
 */
LeaderElectionService getJobManagerLeaderElectionService(JobID jobID);
 /**
 * Gets the leader election service for the cluster's rest endpoint.
 */
default LeaderElectionService getClusterRestEndpointLeaderElectionService()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

public class ZooKeeperLeaderElectionService implements LeaderElectionService, LeaderLatchListener, NodeCacheListener, UnhandledErrorListener {
  // zk 相关
  /** Client to the ZooKeeper quorum. */
	private final CuratorFramework client;

	/** Curator recipe for leader election. */
	private final LeaderLatch leaderLatch;

	/** Curator recipe to watch a given ZooKeeper node for changes. */
	private final NodeCache cache;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93

public class ZooKeeperLeaderRetrievalService implements LeaderRetrievalService, NodeCacheListener, UnhandledErrorListener {
	/** Connection to the used ZooKeeper quorum. */
	private final CuratorFramework client;

	/** Curator recipe to watch changes of a specific ZooKeeper node. */
	private final NodeCache cache;
  
	
	public void start(LeaderRetrievalListener listener) throws Exception {
		synchronized (lock) {
			leaderListener = listener;

			client.getUnhandledErrorListenable().addListener(this);
			cache.getListenable().addListener(this);
			cache.start();

			client.getConnectionStateListenable().addListener(connectionStateListener);

			running = true;
		}
	}

  
	public void stop() throws Exception {
		synchronized (lock) {
			if (!running) {
				return;
			}

			running = false;
		}

		client.getUnhandledErrorListenable().removeListener(this);
		client.getConnectionStateListenable().removeListener(connectionStateListener);

		try {
			cache.close();
		} catch (IOException e) {
			throw new Exception("Could not properly stop the ZooKeeperLeaderRetrievalService.", e);
		}
	}

	
	public void nodeChanged() throws Exception {
		synchronized (lock) {
			if (running) {
				try {
					LOG.debug("Leader node has changed.");

					ChildData childData = cache.getCurrentData();

					String leaderAddress;
					UUID leaderSessionID;

					if (childData == null) {
						leaderAddress = null;
						leaderSessionID = null;
					} else {
						byte[] data = childData.getData();

						if (data == null || data.length == 0) {
							leaderAddress = null;
							leaderSessionID = null;
						} else {
							ByteArrayInputStream bais = new ByteArrayInputStream(data);
							ObjectInputStream ois = new ObjectInputStream(bais);

							leaderAddress = ois.readUTF();
							leaderSessionID = (UUID) ois.readObject();
						}
					}

					if (!(Objects.equals(leaderAddress, lastLeaderAddress) &&
						Objects.equals(leaderSessionID, lastLeaderSessionID))) {
						LOG.debug(
							"New leader information: Leader={}, session ID={}.",
							leaderAddress,
							leaderSessionID);

						lastLeaderAddress = leaderAddress;
						lastLeaderSessionID = leaderSessionID;
						leaderListener.notifyLeaderAddress(leaderAddress, leaderSessionID);
					}
				} catch (Exception e) {
					leaderListener.handleError(new Exception("Could not handle node changed event.", e));
					throw e;
				}
			} else {
				LOG.debug("Ignoring node change notification since the service has already been stopped.");
			}
		}
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

/**
 * A zookeeper based registry for running jobs, highly available.
 */
public class ZooKeeperRunningJobsRegistry implements RunningJobsRegistry {

	private static final Logger LOG = LoggerFactory.getLogger(ZooKeeperRunningJobsRegistry.class);

	private static final Charset ENCODING = Charset.forName("utf-8");

	/** The ZooKeeper client to use. */
	private final CuratorFramework client;

	private final String runningJobPath;

	public ZooKeeperRunningJobsRegistry(final CuratorFramework client, final Configuration configuration) {
		this.client = checkNotNull(client, "client");
		this.runningJobPath = configuration.getString(HighAvailabilityOptions.ZOOKEEPER_RUNNING_JOB_REGISTRY_PATH);
	}

	
	public void setJobRunning(JobID jobID) throws IOException {
		checkNotNull(jobID);

		try {
			writeEnumToZooKeeper(jobID, JobSchedulingStatus.RUNNING);
		}
		catch (Exception e) {
			throw new IOException("Failed to set RUNNING state in ZooKeeper for job " + jobID, e);
		}
	}

	
	public void setJobFinished(JobID jobID) throws IOException {
		checkNotNull(jobID);

		try {
			writeEnumToZooKeeper(jobID, JobSchedulingStatus.DONE);
		}
		catch (Exception e) {
			throw new IOException("Failed to set DONE state in ZooKeeper for job " + jobID, e);
		}
	}

	
	public JobSchedulingStatus getJobSchedulingStatus(JobID jobID) throws IOException {
		checkNotNull(jobID);

		try {
			final String zkPath = createZkPath(jobID);
			final Stat stat = client.checkExists().forPath(zkPath);
			if (stat != null) {
				// found some data, try to parse it
				final byte[] data = client.getData().forPath(zkPath);
				if (data != null) {
					try {
						final String name = new String(data, ENCODING);
						return JobSchedulingStatus.valueOf(name);
					}
					catch (IllegalArgumentException e) {
						throw new IOException("Found corrupt data in ZooKeeper: " +
								Arrays.toString(data) + " is no valid job status");
					}
				}
			}

			// nothing found, yet, must be in status 'PENDING'
			return JobSchedulingStatus.PENDING;
		}
		catch (Exception e) {
			throw new IOException("Get finished state from zk fail for job " + jobID.toString(), e);
		}
	}

	
	public void clearJob(JobID jobID) throws IOException {
		checkNotNull(jobID);

		try {
			final String zkPath = createZkPath(jobID);
			this.client.newNamespaceAwareEnsurePath(zkPath).ensure(client.getZookeeperClient());
			this.client.delete().forPath(zkPath);
		}
		catch (Exception e) {
			throw new IOException("Failed to clear job state from ZooKeeper for job " + jobID, e);
		}
	}

	private String createZkPath(JobID jobID) {
		return runningJobPath + jobID.toString();
	}

	private void writeEnumToZooKeeper(JobID jobID, JobSchedulingStatus status) throws Exception {
		LOG.debug("Setting scheduling state for job {} to {}.", jobID, status);
		final String zkPath = createZkPath(jobID);
		this.client.newNamespaceAwareEnsurePath(zkPath).ensure(client.getZookeeperClient());
		this.client.setData().forPath(zkPath, status.name().getBytes(ENCODING));
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

// Interface offers a common interface for locking on arbitrary
// resources used in leader election.  The Interface is used
// to hide the details on specific implementations in order to allow
// them to change over time.  This interface is strictly for use
// by the leaderelection code.
type Interface interface {
	// Get returns the LeaderElectionRecord
	Get(ctx context.Context) (*LeaderElectionRecord, []byte, error)

	// Create attempts to create a LeaderElectionRecord
	Create(ctx context.Context, ler LeaderElectionRecord) error

	// Update will update and existing LeaderElectionRecord
	Update(ctx context.Context, ler LeaderElectionRecord) error

	// RecordEvent is used to record events
	RecordEvent(string)

	// Identity will return the locks Identity
	Identity() string

	// Describe is used to convert details on current resource lock
	// into a string
	Describe() string
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

// Create attempts to create a LeaderElectionRecord annotation
func (cml *ConfigMapLock) Create(ctx context.Context, ler LeaderElectionRecord) error {
	recordBytes, err := json.Marshal(ler)
	if err != nil {
		return err
	}
	cml.cm, err = cml.Client.ConfigMaps(cml.ConfigMapMeta.Namespace).Create(ctx, &v1.ConfigMap{
		ObjectMeta: metav1.ObjectMeta{
			Name:      cml.ConfigMapMeta.Name,
			Namespace: cml.ConfigMapMeta.Namespace,
			Annotations: map[string]string{
				LeaderElectionRecordAnnotationKey: string(recordBytes),
			},
		},
	}, metav1.CreateOptions{})
	return err
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

 // we use the Lease lock type since edits to Leases are less common
// and fewer objects in the cluster watch "all Leases".
lock := &resourcelock.LeaseLock{
	LeaseMeta: metav1.ObjectMeta{
		Name:      leaseLockName,
		Namespace: leaseLockNamespace,
	},
	Client: client.CoordinationV1(),
	LockConfig: resourcelock.ResourceLockConfig{
		Identity: id,
	},
}

// start the leader election code loop
leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{
	Lock: lock,
	// IMPORTANT: you MUST ensure that any code you have that
	// is protected by the lease must terminate **before**
	// you call cancel. Otherwise, you could have a background
	// loop still running and another process could
	// get elected before your background loop finished, violating
	// the stated goal of the lease.
	ReleaseOnCancel: true,
	LeaseDuration:   60 * time.Second,
	RenewDeadline:   15 * time.Second,
	RetryPeriod:     5 * time.Second,
	Callbacks: leaderelection.LeaderCallbacks{
		OnStartedLeading: func(ctx context.Context) {
			// we're notified when we start - this is where you would
			// usually put your code
			run(ctx)
		},
		OnStoppedLeading: func() {
			// we can do cleanup here
			klog.Infof("leader lost: %s", id)
			os.Exit(0)
		},
		OnNewLeader: func(identity string) {
			// we're notified when new leader elected
			if identity == id {
				// I just got the lock
				return
			}
			klog.Infof("new leader elected: %s", identity)
		},
	},
})

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

/**
   * Runs the leader election in foreground.
   */
  public void run(Runnable startLeadingHook, Runnable stopLeadingHook) {
    run(startLeadingHook, stopLeadingHook, null);
  }

  /**
   * Runs the leader election in foreground.
   */
  public void run(
      Runnable startLeadingHook, Runnable stopLeadingHook, Consumer<String> onNewLeaderHook) {
    this.onNewLeaderHook = onNewLeaderHook;
    log.info("Start leader election with lock {}", config.getLock().describe());
    try {
      if (!acquire()) {
        // Fail to acquire leadership
        return;
      }
      log.info("Successfully acquired lease, became leader");
      // Hook on start leading
      hookWorkers.submit(startLeadingHook);
      renewLoop();
      log.info("Failed to renew lease, lose leadership");
      // Hook on stop leading
      stopLeadingHook.run();
    } catch (Throwable t) {
      stopLeadingHook.run();
    }
  }