Auto-repair on unrecoverable failure
@@ -9,6 +9,7 @@ import com.yahoo.io.IOUtils;
import com.yahoo.vespa.defaults.Defaults;
import io.questdb.cairo.CairoConfiguration;
import io.questdb.cairo.CairoEngine;
+import io.questdb.cairo.CairoException;
import io.questdb.cairo.DefaultCairoConfiguration;
import io.questdb.cairo.TableWriter;
import io.questdb.cairo.sql.Record;
@@ -48,7 +49,7 @@ public class QuestMetricsDb extends AbstractComponent implements MetricsDb {
private final Clock clock;
private final String dataDir;
- private final CairoEngine engine;
+ private CairoEngine engine;
private long highestTimestampAdded = 0;
@@ -64,7 +65,10 @@ public class QuestMetricsDb extends AbstractComponent implements MetricsDb {
&& ! new File(Defaults.getDefaults().vespaHome()).exists())
dataDir = "data"; // We're injected, but not on a node with Vespa installed
this.dataDir = dataDir;
+ initializeDb();
+ }
+ private void initializeDb() {
IOUtils.createDirectory(dataDir + "/" + tableName);
// silence Questdb's custom logging system
@@ -80,6 +84,20 @@ public class QuestMetricsDb extends AbstractComponent implements MetricsDb {
@Override
public void add(Collection<Pair<String, MetricSnapshot>> snapshots) {
try (TableWriter writer = engine.getWriter(newContext().getCairoSecurityContext(), tableName)) {
+ add(snapshots, writer);
+ }
+ catch (CairoException e) {
+ if (e.getMessage().contains("Cannot read offset")) {
+ // This error seems non-recoverable
+ repair(e);
+ try (TableWriter writer = engine.getWriter(newContext().getCairoSecurityContext(), tableName)) {
+ add(snapshots, writer);
+ }
+ }
+ }
+ }
+
+ private void add(Collection<Pair<String, MetricSnapshot>> snapshots, TableWriter writer) {
for (var snapshot : snapshots) {
long atMillis = adjustIfRecent(snapshot.getSecond().at().toEpochMilli(), highestTimestampAdded);
if (atMillis < highestTimestampAdded) continue; // Ignore old data
@@ -96,7 +114,6 @@ public class QuestMetricsDb extends AbstractComponent implements MetricsDb {
}
writer.commit();
}
- }
@Override
public List<NodeTimeseries> getNodeTimeseries(Instant startTime, Set<String> hostnames) {
@@ -152,6 +169,17 @@ public class QuestMetricsDb extends AbstractComponent implements MetricsDb {
engine.close();
}
+ /**
+ * Repairs this db on corruption.
+ *
+ * @param e the exception indicating corruption
+ */
+ private void repair(CairoException e) {
+ log.log(Level.WARNING, "QuestDb seems corrupted, wiping data and starting over", e);
+ IOUtils.recursiveDeleteDir(new File(dataDir));
+ initializeDb();
+ }
+
private void ensureExists(String tableName) {
SqlExecutionContext context = newContext();
if (0 == engine.getStatus(context.getCairoSecurityContext(), new Path(), tableName)) return;
