Skip to content

Commit

Permalink
fix
Browse files Browse the repository at this point in the history 
Signed-off-by: srlch <[email protected]>
  • Loading branch information
@srlch
srlch committed Dec 27, 2024
1 parent e905e7f commit 38fa362
Show file tree
Hide file tree
Showing 6 changed files with 47 additions and 66 deletions.
19 changes: 1 addition & 18 deletions fe/fe-core/src/main/java/com/starrocks/lake/snapshot/ClusterSnapshot.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,19 +15,12 @@
package com.starrocks.lake.snapshot;

import com.google.gson.annotations.SerializedName;
import com.starrocks.common.io.Text;
import com.starrocks.common.io.Writable;
import com.starrocks.persist.gson.GsonUtils;
//import com.starrocks.server.GlobalStateMgr;
//import com.starrocks.server.StorageVolumeMgr;
//import com.starrocks.storagevolume.StorageVolume;
//import com.starrocks.thrift.TClusterSnapshotsItem;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

public class ClusterSnapshot implements Writable {
public class ClusterSnapshot {
public enum ClusterSnapshotType { AUTOMATED }

@SerializedName(value = "snapshotId")
Expand Down Expand Up @@ -68,14 +61,4 @@ public String getSnapshotName() {
public long getSuccessTime() {
return successTime;
}

public static ClusterSnapshot read(DataInput in) throws IOException {
String json = Text.readString(in);
return GsonUtils.GSON.fromJson(json, ClusterSnapshot.class);
}

@Override
public void write(DataOutput out) throws IOException {
Text.writeString(out, GsonUtils.GSON.toJson(this));
}
}
33 changes: 14 additions & 19 deletions fe/fe-core/src/main/java/com/starrocks/lake/snapshot/ClusterSnapshotJob.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,34 +14,39 @@

package com.starrocks.lake.snapshot;

import com.starrocks.common.io.Text;
import com.starrocks.common.io.Writable;
import com.starrocks.persist.gson.GsonUtils;
import com.google.gson.annotations.SerializedName;
import com.starrocks.server.GlobalStateMgr;
//import com.starrocks.thrift.TClusterSnapshotJobsItem;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

public class ClusterSnapshotJob implements Writable {
public class ClusterSnapshotJob {
/*
* INITIALIZING: INIT state for the snapshot.
* SNAPSHOTING: Doing checkpoint/image generation by replaying log for image both for FE and StarMgr and
* then upload the image into remote storage
* FINISHED: Finish backup snapshot
* FINISHED: Finish backup snapshot
* ERROR: End backup snapshot with error
*/
public enum ClusterSnapshotJobState { INITIALIZING, SNAPSHOTING, FINISHED, ERROR }

@SerializedName(value = "jobId")
private long jobId;
@SerializedName(value = "snapshotNamePrefix")
private String snapshotNamePrefix;
@SerializedName(value = "snapshotName")
private String snapshotName;
@SerializedName(value = "storageVolumeName")
private String storageVolumeName;
@SerializedName(value = "createTime")
private long createTime;
@SerializedName(value = "successTime")
private long successTime;
@SerializedName(value = "feJournalId")
private long feJournalId;
@SerializedName(value = "starMgrJournalId")
private long starMgrJournalId;
@SerializedName(value = "state")
private ClusterSnapshotJobState state;
@SerializedName(value = "errMsg")
private String errMsg;

public ClusterSnapshotJob(long jobId, String snapshotNamePrefix, String snapshotName, String storageVolumeName) {
Expand Down Expand Up @@ -117,14 +122,4 @@ public long getJobId() {
private void createSnapshotIfJobIsFinished() {
GlobalStateMgr.getCurrentState().getClusterSnapshotMgr().createAutomatedSnaphot(this);
}

public static ClusterSnapshotJob read(DataInput in) throws IOException {
String json = Text.readString(in);
return GsonUtils.GSON.fromJson(json, ClusterSnapshotJob.class);
}

@Override
public void write(DataOutput out) throws IOException {
Text.writeString(out, GsonUtils.GSON.toJson(this));
}
}
31 changes: 15 additions & 16 deletions fe/fe-core/src/main/java/com/starrocks/lake/snapshot/ClusterSnapshotMgr.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,32 +19,31 @@
import com.starrocks.backup.BlobStorage;
//import com.starrocks.backup.Status;
//import com.starrocks.common.Config;
import com.starrocks.common.io.Text;
import com.starrocks.common.io.Writable;
import com.starrocks.persist.ClusterSnapshotLog;
import com.starrocks.persist.ImageWriter;
//import com.starrocks.persist.Storage;
import com.starrocks.persist.gson.GsonPostProcessable;
import com.starrocks.persist.gson.GsonUtils;
import com.starrocks.persist.metablock.SRMetaBlockEOFException;
import com.starrocks.persist.metablock.SRMetaBlockException;
import com.starrocks.persist.metablock.SRMetaBlockID;
import com.starrocks.persist.metablock.SRMetaBlockReader;
import com.starrocks.persist.metablock.SRMetaBlockWriter;
import com.starrocks.server.GlobalStateMgr;
import com.starrocks.sql.ast.AdminSetAutomatedSnapshotStmt;
import com.starrocks.sql.ast.AdminSetOffAutomatedSnapshotStmt;
import com.starrocks.sql.parser.NodePosition;
import com.starrocks.storagevolume.StorageVolume;
//import com.starrocks.thrift.TClusterSnapshotJobsResponse;
//import com.starrocks.thrift.TClusterSnapshotsResponse;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.io.DataOutput;
//import java.io.File;
import java.io.IOException;
import java.util.List;

// only used for AUTOMATED snapshot for now
public class ClusterSnapshotMgr implements Writable, GsonPostProcessable {
public class ClusterSnapshotMgr implements GsonPostProcessable {
public static final Logger LOG = LogManager.getLogger(ClusterSnapshotMgr.class);
public static final String AUTOMATED_NAME_PREFIX = "automated_cluster_snapshot";
private static final String UPLOAD_SUB_PATH = "/meta/image/";
Expand All @@ -61,7 +60,9 @@ public class ClusterSnapshotMgr implements Writable, GsonPostProcessable {

public ClusterSnapshotMgr() {}

public void addAutomatedSnapshotRequest(String storageVolumeName, boolean isReplay) {
// Turn on automated snapshot, use stmt for extension in future
public void setAutomatedSnapshotOn(AdminSetAutomatedSnapshotStmt stmt, boolean isReplay) {
String storageVolumeName = stmt.getStorageVolumeName();
if (!isReplay) {
ClusterSnapshotLog log = new ClusterSnapshotLog();
log.setCreateSnapshotNamePrefix(AUTOMATED_NAME_PREFIX, storageVolumeName);
Expand Down Expand Up @@ -118,11 +119,11 @@ public boolean containsAutomatedSnapshot() {
return getAutomatedSnapshot() != null;
}

public boolean containAutomatedSnapshotRequest() {
public boolean isAutomatedSnapshotOn() {
return automatedSnapshotSvName != null && !automatedSnapshotSvName.isEmpty();
}

public void addJob(ClusterSnapshotJob job) {
public synchronized void addJob(ClusterSnapshotJob job) {
historyAutomatedSnapshotJobs.add(job);
}

Expand All @@ -134,7 +135,8 @@ public String getLastFinishedAutomatedSnapshotJobName() {
return snapshot.getSnapshotName();
}

public void dropAutomatedSnapshotRequest(boolean isReplay) {
// Turn off automated snapshot, use stmt for extension in future
public void setAutomatedSnapshotOff(AdminSetOffAutomatedSnapshotStmt stmt, boolean isReplay) {
if (!isReplay) {
ClusterSnapshotLog log = new ClusterSnapshotLog();
log.setDropSnapshot(AUTOMATED_NAME_PREFIX);
Expand All @@ -154,7 +156,8 @@ public void replayLog(ClusterSnapshotLog log) {
String createSnapshotNamePrefix = log.getCreateSnapshotNamePrefix();
String storageVolumeName = log.getStorageVolumeName();
if (createSnapshotNamePrefix.equals(AUTOMATED_NAME_PREFIX)) {
addAutomatedSnapshotRequest(storageVolumeName, true);
setAutomatedSnapshotOn(
new AdminSetAutomatedSnapshotStmt(storageVolumeName, NodePosition.ZERO), true);
}
break;
}
Expand All @@ -166,7 +169,8 @@ public void replayLog(ClusterSnapshotLog log) {
case DROP_SNAPSHOT: {
String dropSnapshotName = log.getDropSnapshotName();
if (dropSnapshotName.equals(AUTOMATED_NAME_PREFIX)) {
dropAutomatedSnapshotRequest(true);
setAutomatedSnapshotOff(
new AdminSetOffAutomatedSnapshotStmt(NodePosition.ZERO), true);
}
break;
}
Expand All @@ -187,11 +191,6 @@ public void load(SRMetaBlockReader reader)
ClusterSnapshotMgr data = reader.readJson(ClusterSnapshotMgr.class);
}

@Override
public void write(DataOutput out) throws IOException {
Text.writeString(out, GsonUtils.GSON.toJson(this));
}

@Override
public void gsonPostProcess() throws IOException {
}
Expand Down
5 changes: 2 additions & 3 deletions fe/fe-core/src/main/java/com/starrocks/qe/DDLStmtExecutor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1192,8 +1192,7 @@ public ShowResultSet visitAlterWarehouseStatement(AlterWarehouseStmt stmt, Conne
public ShowResultSet visitAdminSetAutomatedSnapshotStatement(AdminSetAutomatedSnapshotStmt stmt,
ConnectContext context) {
ErrorReport.wrapWithRuntimeException(() -> {
context.getGlobalStateMgr().getClusterSnapshotMgr().addAutomatedSnapshotRequest(
stmt.getStorageVolumeName(), false);
context.getGlobalStateMgr().getClusterSnapshotMgr().setAutomatedSnapshotOn(stmt, false);
});
return null;
}
Expand All @@ -1202,7 +1201,7 @@ public ShowResultSet visitAdminSetAutomatedSnapshotStatement(AdminSetAutomatedSn
public ShowResultSet visitAdminSetOffAutomatedSnapshotStatement(AdminSetOffAutomatedSnapshotStmt stmt,
ConnectContext context) {
ErrorReport.wrapWithRuntimeException(() -> {
context.getGlobalStateMgr().getClusterSnapshotMgr().dropAutomatedSnapshotRequest(false);
context.getGlobalStateMgr().getClusterSnapshotMgr().setAutomatedSnapshotOff(stmt, false);
});
return null;
}
Expand Down
4 changes: 2 additions & 2 deletions fe/fe-core/src/main/java/com/starrocks/sql/analyzer/ClusterSnapshotAnalyzer.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ public static void analyze(StatementBase stmt, ConnectContext session) {
static class ClusterSnapshotAnalyzerVisitor implements AstVisitor<Void, ConnectContext> {
@Override
public Void visitAdminSetAutomatedSnapshotStatement(AdminSetAutomatedSnapshotStmt statement, ConnectContext context) {
if (GlobalStateMgr.getCurrentState().getClusterSnapshotMgr().containAutomatedSnapshotRequest()) {
if (GlobalStateMgr.getCurrentState().getClusterSnapshotMgr().isAutomatedSnapshotOn()) {
throw new SemanticException("Automated snapshot has been turn on");
}

Expand All @@ -52,7 +52,7 @@ public Void visitAdminSetAutomatedSnapshotStatement(AdminSetAutomatedSnapshotStm
@Override
public Void visitAdminSetOffAutomatedSnapshotStatement(AdminSetOffAutomatedSnapshotStmt statement,
ConnectContext context) {
if (!GlobalStateMgr.getCurrentState().getClusterSnapshotMgr().containAutomatedSnapshotRequest()) {
if (!GlobalStateMgr.getCurrentState().getClusterSnapshotMgr().isAutomatedSnapshotOn()) {
throw new SemanticException("Automated snapshot has not been turn on");
}

Expand Down
21 changes: 13 additions & 8 deletions fe/fe-core/src/test/java/com/starrocks/lake/ClusterSnapshotTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,9 @@
//import com.starrocks.server.RunMode;
import com.starrocks.server.StorageVolumeMgr;
import com.starrocks.sql.analyzer.AnalyzeTestUtil;
import com.starrocks.sql.ast.AdminSetAutomatedSnapshotStmt;
import com.starrocks.sql.ast.AdminSetOffAutomatedSnapshotStmt;
import com.starrocks.sql.parser.NodePosition;
import mockit.Delegate;
import mockit.Expectations;
import mockit.Mock;
Expand Down Expand Up @@ -151,15 +154,17 @@ public String getRawServiceId() {
}
};

dropAutomatedSnapshotRequest(false);
setAutomatedSnapshotOff(false);
}

private void addAutomatedSnapshotRequest(boolean testReplay) {
GlobalStateMgr.getCurrentState().getClusterSnapshotMgr().addAutomatedSnapshotRequest(storageVolumeName, testReplay);
private void setAutomatedSnapshotOn(boolean testReplay) {
GlobalStateMgr.getCurrentState().getClusterSnapshotMgr().setAutomatedSnapshotOn(
new AdminSetAutomatedSnapshotStmt(storageVolumeName, NodePosition.ZERO), testReplay);
}

private void dropAutomatedSnapshotRequest(boolean testReplay) {
GlobalStateMgr.getCurrentState().getClusterSnapshotMgr().dropAutomatedSnapshotRequest(testReplay);
private void setAutomatedSnapshotOff(boolean testReplay) {
GlobalStateMgr.getCurrentState().getClusterSnapshotMgr().setAutomatedSnapshotOff(
new AdminSetOffAutomatedSnapshotStmt(NodePosition.ZERO), testReplay);
}

private void initStorageVolume() throws AlreadyExistsException, DdlException, MetaNotFoundException {
Expand All @@ -186,15 +191,15 @@ public void testOperationOfAutomatedSnapshot() throws DdlException {
analyzeFail(turnOnSql + " STORAGE VOLUME testSv");

analyzeSuccess(turnOnSql);
addAutomatedSnapshotRequest(false);
setAutomatedSnapshotOn(false);
// duplicate creation
analyzeFail(turnOnSql);

dropAutomatedSnapshotRequest(false);
setAutomatedSnapshotOff(false);

String turnOFFSql = "ADMIN SET AUTOMATED CLUSTER SNAPSHOT OFF";
analyzeFail(turnOFFSql);
addAutomatedSnapshotRequest(false);
setAutomatedSnapshotOn(false);
analyzeSuccess(turnOFFSql);
}

Expand Down

0 comments on commit 38fa362

Please sign in to comment.