forked from fabric8io/kubernetes-client
-
Notifications
You must be signed in to change notification settings - Fork 0
/
PodUpload.java
142 lines (120 loc) · 5.38 KB
/
PodUpload.java
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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
/**
* Copyright (C) 2015 Red Hat, Inc.
*
* 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 io.fabric8.kubernetes.client.dsl.internal.uploadable;
import io.fabric8.kubernetes.client.dsl.ExecWatch;
import io.fabric8.kubernetes.client.dsl.internal.core.v1.PodOperationsImpl;
import io.fabric8.kubernetes.client.utils.InputStreamPumper;
import io.fabric8.kubernetes.client.utils.Utils;
import io.fabric8.kubernetes.client.utils.internal.Base64;
import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
import org.apache.commons.compress.archivers.tar.TarArchiveOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.zip.GZIPOutputStream;
import static io.fabric8.kubernetes.client.dsl.internal.core.v1.PodOperationsImpl.shellQuote;
public class PodUpload {
private static final String TAR_PATH_DELIMITER = "/";
private PodUpload() {
}
public static boolean upload(PodOperationsImpl operation, Path pathToUpload)
throws IOException {
if (Utils.isNotNullOrEmpty(operation.getContext().getFile()) && pathToUpload.toFile().isFile()) {
return uploadFile(operation, pathToUpload);
} else if (Utils.isNotNullOrEmpty(operation.getContext().getDir()) && pathToUpload.toFile().isDirectory()) {
return uploadDirectory(operation, pathToUpload);
}
throw new IllegalArgumentException("Provided arguments are not valid (file, directory, path)");
}
private static interface UploadProcessor {
void process(OutputStream out) throws IOException;
}
private static boolean upload(PodOperationsImpl operation, String command, UploadProcessor processor) throws IOException {
operation = operation.redirectingInput().terminateOnError();
CompletableFuture<Integer> exitFuture;
try (ExecWatch execWatch = operation.exec("sh", "-c", command)) {
OutputStream out = execWatch.getInput();
processor.process(out);
out.close(); // also flushes
exitFuture = execWatch.exitCode();
}
// TODO: should this timeout be from the start of the upload?
if (!Utils.waitUntilReady(exitFuture, operation.getConfig().getRequestConfig().getUploadRequestTimeout(),
TimeUnit.MILLISECONDS)) {
return false;
}
Integer exitCode = exitFuture.getNow(null);
return exitCode == null || exitCode.intValue() == 0;
}
public static boolean uploadFileData(PodOperationsImpl operation, InputStream inputStream)
throws IOException {
String command = createExecCommandForUpload(operation.getContext().getFile());
return upload(operation, command, os -> {
try (final Base64.InputStream b64In = new Base64.InputStream(inputStream, Base64.ENCODE)) {
InputStreamPumper.transferTo(b64In, os::write);
}
});
}
private static boolean uploadFile(PodOperationsImpl operation, Path pathToUpload)
throws IOException {
try (final FileInputStream fis = new FileInputStream(pathToUpload.toFile())) {
return uploadFileData(operation, fis);
}
}
private static boolean uploadDirectory(PodOperationsImpl operation, Path pathToUpload)
throws IOException {
final String command = String.format(
"mkdir -p %1$s && base64 -d - | tar -C %1$s -xzf -", shellQuote(operation.getContext().getDir()));
return upload(operation, command, os -> {
try (final Base64.OutputStream b64Out = new Base64.OutputStream(os, Base64.ENCODE);
final GZIPOutputStream gzip = new GZIPOutputStream(b64Out);
final TarArchiveOutputStream tar = new TarArchiveOutputStream(gzip)) {
tar.setLongFileMode(TarArchiveOutputStream.LONGFILE_POSIX);
for (File file : pathToUpload.toFile().listFiles()) {
addFileToTar(null, file, tar);
}
tar.flush();
}
});
}
private static void addFileToTar(String rootTarPath, File file, TarArchiveOutputStream tar)
throws IOException {
final String fileName = Optional.ofNullable(rootTarPath).orElse("") + TAR_PATH_DELIMITER + file.getName();
tar.putArchiveEntry(new TarArchiveEntry(file, fileName));
if (file.isFile()) {
Files.copy(file.toPath(), tar);
tar.closeArchiveEntry();
} else if (file.isDirectory()) {
tar.closeArchiveEntry();
for (File fileInDirectory : file.listFiles()) {
addFileToTar(fileName, fileInDirectory, tar);
}
}
}
static String createExecCommandForUpload(String file) {
String directoryTrimmedFromFilePath = file.substring(0, file.lastIndexOf('/'));
final String directory = directoryTrimmedFromFilePath.isEmpty() ? "/" : directoryTrimmedFromFilePath;
return String.format(
"mkdir -p %s && base64 -d - > %s", shellQuote(directory), shellQuote(file));
}
}