feat(dfa): add simple dfa for workflow cluster's status transfer (#3289)

* add fsm

* refactor fsm

* refactor dfa

server/odc-core/src/main/java/com/oceanbase/odc/core/dfa/AbstractDfa.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright (c) 2023 OceanBase.
+ *
+ * Licensed 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 com.oceanbase.odc.core.dfa;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+import org.apache.commons.collections4.CollectionUtils;
+
+import lombok.Getter;
+import lombok.NonNull;
+import lombok.Setter;
+
+/**
+ * {@link AbstractDfa}
+ *
+ * @author yh263208
+ * @date 2024-09-04 20:21
+ * @since ODC_release_4.3.2
+ */
+public abstract class AbstractDfa<STATE, INPUT> {
+
+ @Setter
+ @Getter
+ private STATE currentState;
+ private final List<DfaStateTransfer<STATE, INPUT>> dfaStateTransfers;
+
+ public AbstractDfa(@NonNull List<DfaStateTransfer<STATE, INPUT>> dfaStateTransfers) {
+ this.dfaStateTransfers = dfaStateTransfers;
+ }
+
+ public AbstractDfa<STATE, INPUT> next(INPUT input) throws Exception {
+ if (this.currentState == null) {
+ throw new IllegalStateException("Current state is not set");
+ }
+ List<DfaStateTransfer<STATE, INPUT>> transfers = this.dfaStateTransfers.stream()
+ .filter(t -> t.matchesState(this.currentState)).collect(Collectors.toList());
+ if (CollectionUtils.isEmpty(transfers)) {
+ throw new IllegalStateException("State " + this.currentState + " is the final state");
+ }
+ transfers = transfers.stream().filter(t -> t.matchesInput(input)).collect(Collectors.toList());
+ if (CollectionUtils.isEmpty(transfers)) {
+ throw new IllegalStateException("Unknown input " + input + " for state " + this.currentState);
+ } else if (transfers.size() != 1) {
+ throw new IllegalStateException("More than one routes for state "
+ + this.currentState + " and input " + input);
+ }
+ STATE nextState = transfers.get(0).next();
+ onStateTransfer(currentState, nextState, input);
+ this.currentState = nextState;
+ return this;
+ }
+
+ protected abstract void onStateTransfer(STATE currentState, STATE nextState, INPUT input) throws Exception;
+
+}

server/odc-core/src/main/java/com/oceanbase/odc/core/dfa/DfaStateTransfer.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright (c) 2023 OceanBase.
+ *
+ * Licensed 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 com.oceanbase.odc.core.dfa;
+
+/**
+ * {@link DfaStateTransfer}
+ *
+ * @author yh263208
+ * @date 2024-09-04 20:32
+ * @since ODC_release_4.3.2
+ */
+public interface DfaStateTransfer<STATE, INPUT> {
+
+ STATE next();
+
+ boolean matchesState(STATE state);
+
+ boolean matchesInput(INPUT input);
+
+}

server/odc-core/src/test/java/com/oceanbase/odc/core/dfa/DfaTest.java

@@ -0,0 +1,72 @@
+/*
+ * Copyright (c) 2023 OceanBase.
+ *
+ * Licensed 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 com.oceanbase.odc.core.dfa;
+
+import java.util.Arrays;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Test cases for {@link AbstractDfa}
+ *
+ * @author yh263208
+ * @date 2024-09-04 21:44
+ * @since ODC_release_4.3.2
+ */
+public class DfaTest {
+
+ private static final String FIND_MACHINE = "Find Machine";
+ private static final String POD_CREATED_SUCCEED = "Pod Created Succeed";
+ private static final String IMG_PULL_ERR = "Image pull error";
+ private static final String POD_DELETE = "Pod is deleting";
+ private static final String POD_DELETED = "Pod is deleted";
+
+ @Test
+ public void next_stateExistsRightEvent_nextStateSucceed() throws Exception {
+ AbstractDfa<String, String> fsm = buildFsm();
+ fsm.setCurrentState("Pending");
+ fsm.next(FIND_MACHINE);
+ Assert.assertEquals("Creating", fsm.getCurrentState());
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void next_illegalState_expThrown() throws Exception {
+ AbstractDfa<String, String> fsm = buildFsm();
+ fsm.setCurrentState("Abc");
+ fsm.next(FIND_MACHINE);
+ }
+
+ @Test
+ public void next_reachFinalState_expThrown() throws Exception {
+ AbstractDfa<String, String> fsm = buildFsm();
+ fsm.setCurrentState("Pending");
+ fsm.next(FIND_MACHINE)
+ .next(POD_CREATED_SUCCEED)
+ .next(POD_DELETE)
+ .next(POD_DELETED);
+ }
+
+ private AbstractDfa<String, String> buildFsm() {
+ return new K8sPodStatusDfa(Arrays.asList(
+ new K8sPodStatusDfaTransfer("Pending", "Creating", FIND_MACHINE),
+ new K8sPodStatusDfaTransfer("Creating", "Running", POD_CREATED_SUCCEED),
+ new K8sPodStatusDfaTransfer("Creating", "ImgBackOff", IMG_PULL_ERR),
+ new K8sPodStatusDfaTransfer("Running", "Deleting", POD_DELETE),
+ new K8sPodStatusDfaTransfer("Deleting", "Deleted", POD_DELETED)));
+ }
+
+}

server/odc-core/src/test/java/com/oceanbase/odc/core/dfa/K8sPodStatusDfa.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright (c) 2023 OceanBase.
+ *
+ * Licensed 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 com.oceanbase.odc.core.dfa;
+
+import java.util.List;
+
+import lombok.NonNull;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class K8sPodStatusDfa extends AbstractDfa<String, String> {
+
+ public K8sPodStatusDfa(@NonNull List<DfaStateTransfer<String, String>> dfaStateTransfers) {
+ super(dfaStateTransfers);
+ }
+
+ @Override
+ protected void onStateTransfer(String currentState, String nextState, String input) throws Exception {
+ log.info("Transfer state succeed, currentState={}, nextState={}, event={}", currentState, nextState, input);
+ }
+
+}

server/odc-core/src/test/java/com/oceanbase/odc/core/dfa/K8sPodStatusDfaTransfer.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright (c) 2023 OceanBase.
+ *
+ * Licensed 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 com.oceanbase.odc.core.dfa;
+
+public class K8sPodStatusDfaTransfer implements DfaStateTransfer<String, String> {
+
+ private final String targetState;
+ private final String nextState;
+ private final String targetEvent;
+
+ public K8sPodStatusDfaTransfer(String targetState, String nextState, String targetEvent) {
+ this.targetState = targetState;
+ this.nextState = nextState;
+ this.targetEvent = targetEvent;
+ }
+
+ @Override
+ public String next() {
+ return this.nextState;
+ }
+
+ @Override
+ public boolean matchesState(String s) {
+ return this.targetState.equals(s);
+ }
+
+ @Override
+ public boolean matchesInput(String s) {
+ return this.targetEvent.equals(s);
+ }
+
+}