Dateien aus Cloudflare R2 mit Java und Rclone importieren
Nutzen Sie in Java ProcessBuilder, um rclone copy auszuführen.
Geben Sie das heruntergeladene Verzeichnis erst frei, nachdem rclone erfolgreich beendet wurde.
Das folgende Beispiel importiert ein privates Cloudflare-R2-Präfix, behält verschachtelte Dateinamen
bei und gibt einen Fehlerstatus zurück, wenn die Übertragung fehlschlägt oder ihre Frist überschreitet.
Bucket und Werkzeuge vorbereiten
Diese Anleitung richtet sich an eine Linux-Anwendung mit OpenJDK 21 und einer UTF-8-Locale. Getestet wurde sie mit OpenJDK 21.0.12.1 und rclone 1.75.2. Verwenden Sie eine gepflegte Aktualisierung Ihres JDK; die getestete Patch-Version ist keine grundsätzliche Mindestanforderung. rclone ist ein separates ausführbares Programm, das Sie über die Installationsanleitung erhalten. Java-Abhängigkeiten oder ein Build-Werkzeug sind nicht erforderlich.
Sie benötigen einen vorhandenen R2-Bucket mit den herunterzuladenden Dateien. Lassen Sie dieses
Präfix während des Imports unverändert: Das Kopieren eines sich ändernden Buckets erzeugt keine
Momentaufnahme zu einem bestimmten Zeitpunkt. Verwenden Sie Schlüssel, die unterschiedlichen
Linux-Dateipfaden entsprechen, ohne leere Pfadsegmente oder die Pfadsegmente
. und .., ohne Kollisionen zwischen
Dateien und Verzeichnissen und ohne Namen, die Ihr Dateisystem nicht darstellen kann. Leerzeichen,
verschachtelte Pfade, Unicode und Dateien mit null Bytes werden unterstützt. Beliebige
Objektschlüssel sind nicht zwangsläufig gültige Dateisystempfade.
Prüfen Sie jedes benötigte ausführbare Programm in der Shell, die Java starten wird:
command -v java >/dev/null &&
command -v javac >/dev/null &&
command -v rclone >/dev/null &&
java -version && javac -version && rclone version
Die folgenden Befehle wählen C.UTF-8 aus; diese Locale muss installiert sein.
Arbeiten Sie in einem beschreibbaren Verzeichnis, das Ihnen gehört. Das übergeordnete Verzeichnis
des Ziels muss bereits existieren, atomare Verzeichnisumbenennungen unterstützen und Platz für den
gesamten Download bieten. Führen Sie jeweils nur einen Importer mit exklusivem Zugriff auf dieses
übergeordnete Verzeichnis aus.
Ein privates R2-Remote konfigurieren
Beschaffen Sie sich die Zugriffsschlüssel-ID und den geheimen Zugriffsschlüssel für ein R2-Token, das auf Ihren Bucket beschränkt ist und die Berechtigung Object Read only besitzt. Diese Berechtigung umfasst das Lesen und Auflisten von Objekten über die S3-kompatible API; der Importer muss keine entfernten Objekte erstellen oder löschen. Weitere Informationen finden Sie in Cloudflares Anleitung zur R2-Authentifizierung.
Speichern Sie Folgendes mit einem Texteditor als r2-import.conf in Ihrem
Arbeitsverzeichnis. Ersetzen Sie die drei Platzhalter durch Ihre Zugangsdaten und Ihre Konto-ID.
Dieser Endpunkt gilt für einen Bucket im standardmäßigen Rechtsraum; verwenden Sie gegebenenfalls
den für den Rechtsraum Ihres Buckets spezifischen Endpunkt.
[cloudflare_r2]
type = s3
provider = Cloudflare
access_key_id = YOUR_ACCESS_KEY_ID
secret_access_key = YOUR_SECRET_ACCESS_KEY
endpoint = https://YOUR_ACCOUNT_ID.r2.cloudflarestorage.com
region = auto
no_check_bucket = true
Beschränken Sie den Zugriff auf diese Datei:
chmod 600 ./r2-import.conf
Das Java-Programm übergibt diese Datei explizit mit --config und ist somit
nicht vom üblichen Konfigurationspfad von rclone abhängig. no_check_bucket folgt
Cloudflares Anleitung zur rclone-Konfiguration für Tokens mit
Objektberechtigungen. Committen Sie die Datei nicht und fügen Sie keine Zugangsdaten in
Befehlsargumente ein. Die vom Kindprozess geerbten Variablen RCLONE_*
werden entfernt, damit Einstellungen wie ein globaler Probelauf diesen Import nicht unbemerkt
in einen anderen Vorgang verwandeln können.
Den Java-Importer speichern
Speichern Sie diese vollständige Datei als RcloneImporter.java. Die Argumente sind das
ausführbare rclone-Programm, die Konfigurationsdatei, das Remote-Präfix, das neue Zielverzeichnis und
das Zeitlimit für die gesamte Übertragung in Sekunden. Verwenden Sie vertrauenswürdige
Anwendungskonfiguration, statt beliebige Programmpfade oder Remotes von Nutzern zu akzeptieren.
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.FileVisitResult;
import java.nio.file.Files;
import java.nio.file.LinkOption;
import java.nio.file.Path;
import java.nio.file.SimpleFileVisitor;
import java.nio.file.StandardCopyOption;
import java.nio.file.attribute.BasicFileAttributes;
import java.time.Duration;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
public class RcloneImporter {
private static final int DIAGNOSTIC_BYTES = 8192;
private static void runRclone(ProcessBuilder builder, Duration timeout)
throws IOException, InterruptedException {
Process process = builder.redirectErrorStream(true).start();
var diagnostics = new ByteArrayOutputStream();
var readFailure = new AtomicReference<IOException>();
Thread drain = Thread.ofVirtual().start(() -> {
try (var input = process.getInputStream()) {
byte[] buffer = new byte[4096];
int count;
while ((count = input.read(buffer)) != -1) {
int kept = Math.min(count, DIAGNOSTIC_BYTES - diagnostics.size());
if (kept > 0) diagnostics.write(buffer, 0, kept);
}
} catch (IOException error) {
readFailure.set(error);
}
});
try {
if (!process.waitFor(timeout.toMillis(), TimeUnit.MILLISECONDS)) {
throw new IOException("rclone timed out after " + timeout);
}
drain.join();
if (readFailure.get() != null) throw readFailure.get();
System.err.print(diagnostics.toString(StandardCharsets.UTF_8));
if (process.exitValue() != 0) {
throw new IOException("rclone exited with code " + process.exitValue());
}
} finally {
// Reap the direct rclone child before deleting anything it might still be writing.
process.destroyForcibly();
boolean interrupted = false;
while (process.isAlive() || drain.isAlive()) {
try {
process.waitFor();
drain.join();
} catch (InterruptedException error) {
interrupted = true;
}
}
if (interrupted) Thread.currentThread().interrupt();
}
}
private static void removeStaging(Path staging) throws IOException {
Files.walkFileTree(staging, new SimpleFileVisitor<Path>() {
@Override
public FileVisitResult visitFile(Path file, BasicFileAttributes attributes)
throws IOException {
Files.delete(file);
return FileVisitResult.CONTINUE;
}
@Override
public FileVisitResult postVisitDirectory(Path directory, IOException error)
throws IOException {
if (error != null) throw error;
Files.delete(directory);
return FileVisitResult.CONTINUE;
}
});
}
public static void importFiles(Path executable, Path config, String source, Path destination,
Duration timeout) throws IOException, InterruptedException {
if (!Files.isExecutable(executable) || !Files.isRegularFile(config)) {
throw new IOException("Check the rclone executable and configuration file");
}
if (!source.startsWith("cloudflare_r2:") || !source.endsWith("/")) {
throw new IllegalArgumentException("Use cloudflare_r2:bucket/prefix/ as the source");
}
if (timeout.isNegative() || timeout.isZero() || timeout.compareTo(Duration.ofDays(1)) > 0) {
throw new IllegalArgumentException("Timeout must be between 1 and 86400 seconds");
}
Path name = destination.getFileName();
if (name == null || name.toString().equals(".") || name.toString().equals("..")) {
throw new IllegalArgumentException("Choose a new destination directory name");
}
Path target = destination.toAbsolutePath();
if (Files.exists(target, LinkOption.NOFOLLOW_LINKS)) {
throw new IOException("Destination already exists; choose a new directory");
}
Path staging = Files.createTempDirectory(target.getParent(), ".r2-import-");
boolean published = false;
try {
ProcessBuilder builder = new ProcessBuilder(executable.toAbsolutePath().toString(),
"--config", config.toAbsolutePath().toString(), "copy",
"--", source, staging.toString());
builder.environment().keySet().removeIf(key -> key.startsWith("RCLONE_"));
runRclone(builder, timeout);
if (Thread.currentThread().isInterrupted()) throw new InterruptedException();
Files.move(staging, target, StandardCopyOption.ATOMIC_MOVE);
published = true;
} finally {
if (!published) removeStaging(staging);
}
}
public static void main(String[] args) {
if (args.length != 5) {
System.err.println("Usage: java RcloneImporter RCLONE CONFIG REMOTE DEST SECONDS");
System.exit(2);
}
try {
importFiles(Path.of(args[0]), Path.of(args[1]), args[2], Path.of(args[3]),
Duration.ofSeconds(Long.parseLong(args[4])));
System.out.println("Import complete: " + args[3]);
} catch (InterruptedException error) {
Thread.currentThread().interrupt();
System.err.println("Import interrupted; destination was not published");
System.exit(130);
} catch (IOException | IllegalArgumentException error) {
System.err.println("Import failed: " + error.getMessage());
System.exit(1);
}
}
}
Beide Ausgabeströme des Kindprozesses fließen in eine gemeinsame Pipe. Ein separater virtueller
Thread liest diese Pipe kontinuierlich aus und behält nur die ersten 8 KiB für die Diagnose im
Terminal. So kann der aufrufende Thread die Übertragungsfrist durchsetzen, selbst wenn rclone mehr
Ausgabe erzeugt, als eine Pipe aufnehmen kann. Die Java-Dokumentation zu
Process erläutert das Risiko blockierter
Ausgaben von Unterprozessen.
Das Staging-Verzeichnis liegt im selben übergeordneten Verzeichnis wie das endgültige Ziel. Nachdem rclone mit dem Exit-Status null beendet wurde, fordert Java eine atomare Verschiebung an. Kann das Dateisystem sie nicht bereitstellen, schlägt der Import fehl. Atomare Verschiebungen sind keine Überschreibsperre: Ein anderer schreibender Prozess kann zwischen Existenzprüfung und Verschiebung eingreifen. Deshalb erfordert dieses Beispiel exklusiven Zugriff.
Kompilieren und das Präfix importieren
Ersetzen Sie /absolute/path/to/rclone durch das zuvor gefundene ausführbare Programm und
verwenden Sie Ihren Bucket und Ihr Präfix. Behalten Sie die Anführungszeichen um Pfade bei. Führen
Sie die beiden Befehle zusammen aus, damit bei einem Kompilierungsfehler keine ältere Klassendatei
gestartet werden kann:
javac -proc:none -encoding UTF-8 RcloneImporter.java &&
LC_ALL=C.UTF-8 java -cp . RcloneImporter /absolute/path/to/rclone ./r2-import.conf \
'cloudflare_r2:my-data-bucket/source-files/' ./downloaded-files 600
Bei Erfolg gibt das Programm Import complete: ./downloaded-files aus und beendet sich mit dem Status null.
Aus einem Objekt wie source-files/reports/July résumé.txt wird downloaded-files/reports/July résumé.txt.
copy kopiert den Inhalt des Präfixes,
ohne source-files als zusätzliches Verzeichnis anzulegen. Die entfernten Objekte
werden dabei nicht gelöscht.
Die Frist von 600 Sekunden umfasst auch die Retries von rclone. Deshalb kann selbst eine gültige Datei bei einer langsamen Verbindung das Zeitlimit überschreiten. Bei einer Zeitüberschreitung wird der direkte rclone-Kindprozess beendet und dessen Ende abgewartet. Anschließend wird das Staging-Verzeichnis dieses Aufrufs entfernt; das endgültige Ziel bleibt abwesend. Wählen Sie ein längeres Zeitlimit von bis zu 86.400 Sekunden und versuchen Sie es mit demselben Ziel erneut, nachdem der fehlgeschlagene Prozess beendet wurde. Ein vorhandenes Ziel, auch ein symbolischer Link, wird abgelehnt; verwenden Sie für einen weiteren abgeschlossenen Import einen neuen Namen.
Einen fehlgeschlagenen Import diagnostizieren
- Beendigung des Kindprozesses: Lesen Sie die begrenzte Diagnoseausgabe im Terminal und prüfen Sie dann die gewählte Konfigurationsdatei, die Bucket-Beschränkung des Tokens, den Endpunkt und den ausgehenden HTTPS-Zugriff. Eine fehlgeschlagene Anfrage ist ein Fehler und kein erfolgreicher Download ohne Inhalt.
- Keine Dateien: Ein leeres oder nicht vorhandenes Präfix kann erfolgreich als leeres Verzeichnis kopiert werden. Vergewissern Sie sich, dass das Präfix stimmt, und prüfen Sie das Ergebnis vor der Verwendung; der Exit-Status null beweist nicht, dass ein bestimmtes Objekt existierte. Halten Sie bei Bedarf ein unabhängiges Sollverzeichnis oder Prüfsummen des Herausgebers bereit.
- Unterbrechung: Eine Anwendung, die
importFilesaufruft, kann ihren wartenden Java-Thread unterbrechen; die Methode stoppt den Kindprozess und bereinigt das Staging-Verzeichnis, bevor sieInterruptedExceptionweitergibt. Der Aufrufer ist dafür verantwortlich, sein Interrupt-Flag wiederherzustellen. Dies ist etwas anderes als das Beenden der JVM. - Abrupte Beendigung: Stromausfall,
kill -9und die Beendigung der JVM können Verzeichnisse mit dem Namen.r2-import-*oder einen verwaisten Kindprozess hinterlassen. Stoppen Sie jeden verbliebenen Kindprozess, bevor Sie ausschließlich das Staging-Verzeichnis dieses Aufrufs entfernen. Diese Verzeichnisse enthalten unvollständige Arbeitsergebnisse und sind niemals das endgültige Ziel. Auch die Bereinigung kann fehlschlagen, wenn sich Dateisystemberechtigungen oder die Verfügbarkeit des Dateisystems ändern.
Den Ausführungsort für Importe wählen
Dieser Ansatz eignet sich für einen Java-Dienst, der bereits rclone verwendet und lokale Dateien benötigt, bevor er sie verarbeitet. Sollen die Dateien stattdessen in einen gehosteten Transloadit-Workflow gelangen, finden Sie weitere Informationen beim Cloudflare Import Robot und im Java SDK.