Skip to content

Commit 7c483d1

Browse files
tutengtuteng
authored
Support mount secret file, fixed namespace (#158)
* Support mount secret file * Fixed k8s namespace * Upgrade pulsar broker version * Update pulsar groupId
1 parent 66e7b97 commit 7c483d1

9 files changed

Lines changed: 177 additions & 135 deletions

File tree

‎mesh-worker-service/pom.xml‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@
2929
<version>v0.1.5-SNAPSHOT</version>
3030

3131
<properties>
32-
<pulsar.version>2.8.0-rc-202103292206</pulsar.version>
32+
<pulsar.version>2.8.0-rc-202105140121</pulsar.version>
3333
<lombok.version>1.18.16</lombok.version>
3434
<log4j2.version>2.14.0</log4j2.version>
3535
<kubernetes-client.version>10.0.1</kubernetes-client.version>
@@ -42,7 +42,7 @@
4242

4343
<dependencies>
4444
<dependency>
45-
<groupId>org.apache.pulsar</groupId>
45+
<groupId>io.streamnative</groupId>
4646
<artifactId>pulsar-functions-worker</artifactId>
4747
<version>${pulsar.version}</version>
4848
</dependency>

‎mesh-worker-service/src/main/java/io/functionmesh/compute/MeshWorkerService.java‎

Lines changed: 0 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,6 @@
2828
import io.kubernetes.client.util.Config;
2929
import lombok.Getter;
3030
import lombok.extern.slf4j.Slf4j;
31-
import org.apache.commons.lang3.StringUtils;
3231
import org.apache.pulsar.broker.ServiceConfiguration;
3332
import org.apache.pulsar.broker.authentication.AuthenticationService;
3433
import org.apache.pulsar.broker.authorization.AuthorizationService;
@@ -38,11 +37,8 @@
3837
import org.apache.pulsar.client.api.PulsarClient;
3938
import org.apache.pulsar.common.conf.InternalConfigurationData;
4039
import org.apache.pulsar.common.util.SimpleTextOutputStream;
41-
import org.apache.pulsar.functions.auth.FunctionAuthProvider;
42-
import org.apache.pulsar.functions.auth.KubernetesFunctionAuthProvider;
4340
import org.apache.pulsar.functions.runtime.RuntimeUtils;
4441
import org.apache.pulsar.functions.runtime.kubernetes.KubernetesRuntimeFactoryConfig;
45-
import org.apache.pulsar.functions.worker.ConnectorsManager;
4642
import org.apache.pulsar.functions.worker.ErrorNotifier;
4743
import org.apache.pulsar.functions.worker.PulsarWorkerService;
4844
import org.apache.pulsar.functions.worker.WorkerConfig;
@@ -76,7 +72,6 @@ public class MeshWorkerService implements WorkerService {
7672
private CustomObjectsApi customObjectsApi;
7773
private ApiClient apiClient;
7874
private PulsarAdmin brokerAdmin;
79-
private Optional<KubernetesFunctionAuthProvider> authProvider;
8075
private KubernetesRuntimeFactoryConfig factoryConfig;
8176

8277
private AuthenticationService authenticationService;
@@ -138,13 +133,6 @@ public void init(WorkerConfig workerConfig) throws Exception {
138133
this.workerConfig = workerConfig;
139134
this.initKubernetesClient();
140135
this.authenticationEnabled = this.workerConfig.isAuthenticationEnabled();
141-
if (this.workerConfig.isAuthenticationEnabled() && !StringUtils.isEmpty(this.workerConfig.getFunctionAuthProviderClassName())) {
142-
Optional<FunctionAuthProvider> functionAuthProvider = Optional.empty();
143-
functionAuthProvider = Optional.of(FunctionAuthProvider.getAuthProvider(workerConfig.getFunctionAuthProviderClassName()));
144-
KubernetesFunctionAuthProvider kubernetesFunctionAuthProvider = (KubernetesFunctionAuthProvider) functionAuthProvider.get();
145-
kubernetesFunctionAuthProvider.initialize(coreV1Api, null, null);
146-
this.authProvider = Optional.of(kubernetesFunctionAuthProvider);
147-
}
148136
this.functions = new FunctionsImpl(() -> MeshWorkerService.this);
149137
this.sources = new SourcesImpl(() -> MeshWorkerService.this);
150138
this.sinks = new SinksImpl(() -> MeshWorkerService.this);

‎mesh-worker-service/src/main/java/io/functionmesh/compute/rest/api/FunctionsImpl.java‎

Lines changed: 52 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,8 @@
2020

2121
import com.google.common.collect.Maps;
2222
import io.functionmesh.compute.functions.models.V1alpha1FunctionSpecPod;
23+
import io.functionmesh.compute.functions.models.V1alpha1FunctionSpecPodVolumeMounts;
24+
import io.functionmesh.compute.functions.models.V1alpha1FunctionSpecPodVolumes;
2325
import io.functionmesh.compute.util.FunctionsUtil;
2426
import io.functionmesh.compute.functions.models.V1alpha1Function;
2527
import io.functionmesh.compute.MeshWorkerService;
@@ -36,10 +38,10 @@
3638
import org.apache.pulsar.functions.utils.ComponentTypeUtils;
3739
import org.apache.pulsar.functions.worker.service.api.Functions;
3840
import org.glassfish.jersey.media.multipart.FormDataContentDisposition;
39-
4041
import javax.ws.rs.core.Response;
4142
import java.io.InputStream;
4243
import java.net.URI;
44+
import java.util.List;
4345
import java.util.Map;
4446
import java.util.function.Supplier;
4547

@@ -102,6 +104,8 @@ public void registerFunction(final String tenant,
102104
functionPkgUrl,
103105
functionConfig
104106
);
107+
// override namespace by configuration file
108+
v1alpha1Function.getMetadata().setNamespace(KubernetesUtils.getNamespace(worker().getFactoryConfig()));
105109
Map<String, String> customLabels = Maps.newHashMap();
106110
customLabels.put(TENANT_LABEL_CLAIM, tenant);
107111
customLabels.put(NAMESPACE_LABEL_CLAIM, namespace);
@@ -112,35 +116,7 @@ public void registerFunction(final String tenant,
112116
pod.setLabels(customLabels);
113117
v1alpha1Function.getSpec().setPod(pod);
114118
try {
115-
if (worker().getWorkerConfig().isAuthenticationEnabled()) {
116-
Function.FunctionDetails.Builder functionDetailsBuilder = Function.FunctionDetails.newBuilder();
117-
functionDetailsBuilder.setTenant(tenant);
118-
functionDetailsBuilder.setNamespace(namespace);
119-
functionDetailsBuilder.setName(functionName);
120-
worker().getAuthProvider().ifPresent(functionAuthProvider -> {
121-
if (clientAuthenticationDataHttps != null) {
122-
try {
123-
String authSecretName = KubernetesUtils.upsertSecret(kind.toLowerCase(), "auth",
124-
v1alpha1Function.getSpec().getClusterName(), tenant, namespace, functionName,
125-
worker().getWorkerConfig(), worker().getCoreV1Api(), worker().getFactoryConfig());
126-
v1alpha1Function.getSpec().getPulsar().setAuthSecret(authSecretName);
127-
String tlsSecretName = KubernetesUtils.upsertSecret(kind.toLowerCase(), "tls",
128-
v1alpha1Function.getSpec().getClusterName(), tenant, namespace, functionName,
129-
worker().getWorkerConfig(), worker().getCoreV1Api(), worker().getFactoryConfig());
130-
v1alpha1Function.getSpec().getPulsar().setTlsSecret(tlsSecretName);
131-
} catch (Exception e) {
132-
log.error("Error caching authentication data for {} {}/{}/{}",
133-
ComponentTypeUtils.toString(componentType), tenant, namespace, functionName, e);
134-
135-
136-
throw new RestException(
137-
Response.Status.INTERNAL_SERVER_ERROR,
138-
String.format("Error caching authentication data for %s %s:- %s",
139-
ComponentTypeUtils.toString(componentType), functionName, e.getMessage()));
140-
}
141-
}
142-
});
143-
}
119+
this.upsertFunction(tenant, namespace, functionName, functionConfig, v1alpha1Function, clientAuthenticationDataHttps);
144120
Call call = worker().getCustomObjectsApi().createNamespacedCustomObjectCall(
145121
group,
146122
version,
@@ -191,6 +167,7 @@ public void updateFunction(final String tenant,
191167
functionConfig
192168
);
193169
v1alpha1Function.getMetadata().setResourceVersion(oldFn.getMetadata().getResourceVersion());
170+
this.upsertFunction(tenant, namespace, functionName, functionConfig, v1alpha1Function, clientAuthenticationDataHttps);
194171
Call replaceCall = worker().getCustomObjectsApi().replaceNamespacedCustomObjectCall(
195172
group,
196173
version,
@@ -304,7 +281,51 @@ public void updateFunctionOnWorkerLeader(final String tenant,
304281
final InputStream uploadedInputStream,
305282
final boolean delete,
306283
URI uri,
307-
final String clientRole) {
284+
final String clientRole,
285+
final AuthenticationDataSource clientAuthenticationDataHttps) {
286+
287+
}
308288

289+
private void upsertFunction(final String tenant,
290+
final String namespace,
291+
final String functionName,
292+
final FunctionConfig functionConfig,
293+
V1alpha1Function v1alpha1Function,
294+
AuthenticationDataHttps clientAuthenticationDataHttps) {
295+
if (worker().getWorkerConfig().isAuthenticationEnabled()) {
296+
if (clientAuthenticationDataHttps != null) {
297+
try {
298+
299+
Map<String, Object> functionsWorkerServiceCustomConfigs = worker()
300+
.getWorkerConfig().getFunctionsWorkerServiceCustomConfigs();
301+
Object volumes = functionsWorkerServiceCustomConfigs.get("volumes");
302+
if (volumes != null) {
303+
List<V1alpha1FunctionSpecPodVolumes> volumesList = (List<V1alpha1FunctionSpecPodVolumes>) volumes;
304+
v1alpha1Function.getSpec().getPod().setVolumes(volumesList);
305+
}
306+
Object volumeMounts = functionsWorkerServiceCustomConfigs.get("volumeMounts");
307+
if (volumeMounts != null) {
308+
List<V1alpha1FunctionSpecPodVolumeMounts> volumeMountsList = (List<V1alpha1FunctionSpecPodVolumeMounts>) volumeMounts;
309+
v1alpha1Function.getSpec().setVolumeMounts(volumeMountsList);
310+
}
311+
String authSecretName = KubernetesUtils.upsertSecret(kind.toLowerCase(), "auth",
312+
v1alpha1Function.getSpec().getClusterName(), tenant, namespace, functionName,
313+
worker().getWorkerConfig(), worker().getCoreV1Api(), worker().getFactoryConfig());
314+
v1alpha1Function.getSpec().getPulsar().setAuthSecret(authSecretName);
315+
String tlsSecretName = KubernetesUtils.upsertSecret(kind.toLowerCase(), "tls",
316+
v1alpha1Function.getSpec().getClusterName(), tenant, namespace, functionName,
317+
worker().getWorkerConfig(), worker().getCoreV1Api(), worker().getFactoryConfig());
318+
v1alpha1Function.getSpec().getPulsar().setTlsSecret(tlsSecretName);
319+
} catch (Exception e) {
320+
log.error("Error create or update auth or tls secret for {} {}/{}/{}",
321+
ComponentTypeUtils.toString(componentType), tenant, namespace, functionName, e);
322+
323+
324+
throw new RestException(Response.Status.INTERNAL_SERVER_ERROR,
325+
String.format("Error create or update auth or tls secret for %s %s:- %s",
326+
ComponentTypeUtils.toString(componentType), functionName, e.getMessage()));
327+
}
328+
}
329+
}
309330
}
310331
}

‎mesh-worker-service/src/main/java/io/functionmesh/compute/rest/api/MeshComponentImpl.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -348,7 +348,8 @@ public void putFunctionState(final String tenant,
348348
@Override
349349
public void uploadFunction(final InputStream uploadedInputStream,
350350
final String path,
351-
String clientRole) {
351+
String clientRole,
352+
final AuthenticationDataSource clientAuthenticationDataHttps) {
352353

353354
}
354355

@@ -376,7 +377,7 @@ public List<ConnectorDefinition> getListOfConnectors() {
376377
}
377378

378379
@Override
379-
public void reloadConnectors(String clientRole) {
380+
public void reloadConnectors(String clientRole, final AuthenticationDataSource clientAuthenticationDataHttps) {
380381
meshWorkerServiceSupplier.get().getConnectorsManager().reloadConnectors();
381382
}
382383

‎mesh-worker-service/src/main/java/io/functionmesh/compute/rest/api/SinksImpl.java‎

Lines changed: 48 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,8 @@
2121
import com.google.common.collect.Maps;
2222
import io.functionmesh.compute.sinks.models.V1alpha1Sink;
2323
import io.functionmesh.compute.sinks.models.V1alpha1SinkSpecPod;
24+
import io.functionmesh.compute.sinks.models.V1alpha1SinkSpecPodVolumeMounts;
25+
import io.functionmesh.compute.sinks.models.V1alpha1SinkSpecPodVolumes;
2426
import io.functionmesh.compute.util.KubernetesUtils;
2527
import io.functionmesh.compute.util.SinksUtil;
2628
import io.functionmesh.compute.MeshWorkerService;
@@ -40,7 +42,6 @@
4042
import org.apache.pulsar.functions.utils.ComponentTypeUtils;
4143
import org.apache.pulsar.functions.worker.service.api.Sinks;
4244
import org.glassfish.jersey.media.multipart.FormDataContentDisposition;
43-
4445
import javax.ws.rs.core.Response;
4546
import java.io.InputStream;
4647
import java.net.URI;
@@ -101,8 +102,7 @@ public void registerSink(
101102
clientAuthenticationDataHttps,
102103
ComponentTypeUtils.toString(componentType));
103104
this.validateTenantIsExist(tenant, namespace, sinkName, clientRole);
104-
V1alpha1Sink v1alpha1Sink;
105-
v1alpha1Sink =
105+
V1alpha1Sink v1alpha1Sink =
106106
SinksUtil.createV1alpha1SkinFromSinkConfig(
107107
kind,
108108
group,
@@ -112,6 +112,7 @@ public void registerSink(
112112
uploadedInputStream,
113113
sinkConfig,
114114
this.meshWorkerServiceSupplier.get().getConnectorsManager());
115+
// override namesapce by configuration
115116
v1alpha1Sink.getMetadata().setNamespace(KubernetesUtils.getNamespace(worker().getFactoryConfig()));
116117
try {
117118
Map<String, String> customLabels = Maps.newHashMap();
@@ -125,33 +126,7 @@ public void registerSink(
125126
}
126127
pod.setLabels(customLabels);
127128
v1alpha1Sink.getSpec().setPod(pod);
128-
if (worker().getWorkerConfig().isAuthenticationEnabled()) {
129-
Function.FunctionDetails.Builder functionDetailsBuilder = Function.FunctionDetails.newBuilder();
130-
functionDetailsBuilder.setTenant(tenant);
131-
functionDetailsBuilder.setNamespace(namespace);
132-
functionDetailsBuilder.setName(sinkName);
133-
worker().getAuthProvider().ifPresent(functionAuthProvider -> {
134-
if (clientAuthenticationDataHttps != null) {
135-
try {
136-
String authSecretName = KubernetesUtils.upsertSecret(kind.toLowerCase(), "auth",
137-
v1alpha1Sink.getSpec().getClusterName(), tenant, namespace, sinkName,
138-
worker().getWorkerConfig(), worker().getCoreV1Api(), worker().getFactoryConfig());
139-
v1alpha1Sink.getSpec().getPulsar().setAuthSecret(authSecretName);
140-
String tlsSecretName = KubernetesUtils.upsertSecret(kind.toLowerCase(), "tls",
141-
v1alpha1Sink.getSpec().getClusterName(), tenant, namespace, sinkName,
142-
worker().getWorkerConfig(), worker().getCoreV1Api(), worker().getFactoryConfig());
143-
v1alpha1Sink.getSpec().getPulsar().setTlsSecret(tlsSecretName);
144-
} catch (Exception e) {
145-
log.error("Error caching authentication data for {} {}/{}/{}",
146-
ComponentTypeUtils.toString(componentType), tenant, namespace, sinkName, e);
147-
148-
149-
throw new RestException(Response.Status.INTERNAL_SERVER_ERROR, String.format("Error caching authentication data for %s %s:- %s",
150-
ComponentTypeUtils.toString(componentType), sinkName, e.getMessage()));
151-
}
152-
}
153-
});
154-
}
129+
this.upsertSink(tenant, namespace, sinkName, sinkConfig, v1alpha1Sink, clientAuthenticationDataHttps);
155130
Call call =
156131
worker().getCustomObjectsApi()
157132
.createNamespacedCustomObjectCall(
@@ -211,6 +186,7 @@ public void updateSink(
211186
sinkPkgUrl,
212187
uploadedInputStream,
213188
sinkConfig, this.meshWorkerServiceSupplier.get().getConnectorsManager());
189+
this.upsertSink(tenant, namespace, sinkName, sinkConfig, v1alpha1Sink, clientAuthenticationDataHttps);
214190
v1alpha1Sink.getMetadata().setNamespace(KubernetesUtils.getNamespace(worker().getFactoryConfig()));
215191
v1alpha1Sink
216192
.getMetadata()
@@ -355,4 +331,46 @@ public List<ConnectorDefinition> getSinkList() {
355331
public List<ConfigFieldDefinition> getSinkConfigDefinition(String name) {
356332
return new ArrayList<>();
357333
}
334+
335+
private void upsertSink(final String tenant,
336+
final String namespace,
337+
final String sinkName,
338+
final SinkConfig sinkConfig,
339+
V1alpha1Sink v1alpha1Sink,
340+
AuthenticationDataHttps clientAuthenticationDataHttps) {
341+
if (worker().getWorkerConfig().isAuthenticationEnabled()) {
342+
if (clientAuthenticationDataHttps != null) {
343+
try {
344+
Map<String, Object> functionsWorkerServiceCustomConfigs = worker()
345+
.getWorkerConfig().getFunctionsWorkerServiceCustomConfigs();
346+
Object volumes = functionsWorkerServiceCustomConfigs.get("volumes");
347+
if (volumes != null) {
348+
List<V1alpha1SinkSpecPodVolumes> volumesList = (List<V1alpha1SinkSpecPodVolumes>) volumes;
349+
v1alpha1Sink.getSpec().getPod().setVolumes(volumesList);
350+
}
351+
Object volumeMounts = functionsWorkerServiceCustomConfigs.get("volumeMounts");
352+
if (volumeMounts != null) {
353+
List<V1alpha1SinkSpecPodVolumeMounts> volumeMountsList = (List<V1alpha1SinkSpecPodVolumeMounts>) volumeMounts;
354+
v1alpha1Sink.getSpec().setVolumeMounts(volumeMountsList);
355+
}
356+
String authSecretName = KubernetesUtils.upsertSecret(kind.toLowerCase(), "auth",
357+
v1alpha1Sink.getSpec().getClusterName(), tenant, namespace, sinkName,
358+
worker().getWorkerConfig(), worker().getCoreV1Api(), worker().getFactoryConfig());
359+
v1alpha1Sink.getSpec().getPulsar().setAuthSecret(authSecretName);
360+
String tlsSecretName = KubernetesUtils.upsertSecret(kind.toLowerCase(), "tls",
361+
v1alpha1Sink.getSpec().getClusterName(), tenant, namespace, sinkName,
362+
worker().getWorkerConfig(), worker().getCoreV1Api(), worker().getFactoryConfig());
363+
v1alpha1Sink.getSpec().getPulsar().setTlsSecret(tlsSecretName);
364+
} catch (Exception e) {
365+
log.error("Error create or update auth or tls secret data for {} {}/{}/{}",
366+
ComponentTypeUtils.toString(componentType), tenant, namespace, sinkName, e);
367+
368+
369+
throw new RestException(Response.Status.INTERNAL_SERVER_ERROR,
370+
String.format("Error create or update auth or tls secret for %s %s:- %s",
371+
ComponentTypeUtils.toString(componentType), sinkName, e.getMessage()));
372+
}
373+
}
374+
}
375+
}
358376
}

0 commit comments

Comments
 (0)