Skip to content

Commit c17be14

Browse files
tutengtuteng
authored
Support filter tenant and namespace (#176)
* Support filter tenant and namespace * Support get by hash name
1 parent 7d4383a commit c17be14

File tree

14 files changed

+291
-186
lines changed

14 files changed

+291
-186
lines changed

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

Lines changed: 31 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import io.functionmesh.compute.functions.models.V1alpha1FunctionSpecPod;
2424
import io.functionmesh.compute.functions.models.V1alpha1FunctionSpecPodVolumeMounts;
2525
import io.functionmesh.compute.functions.models.V1alpha1FunctionSpecPodVolumes;
26+
import io.functionmesh.compute.util.CommonUtil;
2627
import io.functionmesh.compute.util.FunctionsUtil;
2728
import io.functionmesh.compute.functions.models.V1alpha1Function;
2829
import io.functionmesh.compute.MeshWorkerService;
@@ -118,26 +119,31 @@ public void registerFunction(final String tenant,
118119
ComponentTypeUtils.toString(componentType));
119120
this.validateTenantIsExist(tenant, namespace, functionName, clientRole);
120121

122+
String cluster = worker().getWorkerConfig().getPulsarFunctionsCluster();
121123
V1alpha1Function v1alpha1Function = FunctionsUtil.createV1alpha1FunctionFromFunctionConfig(
122124
kind,
123125
group,
124126
version,
125127
functionName,
126128
functionPkgUrl,
127129
functionConfig,
128-
worker().getWorkerConfig().getFunctionsWorkerServiceCustomConfigs()
130+
worker().getWorkerConfig().getFunctionsWorkerServiceCustomConfigs(),
131+
cluster
129132
);
130133
// override namespace by configuration file
131134
v1alpha1Function.getMetadata().setNamespace(KubernetesUtils.getNamespace(worker().getFactoryConfig()));
132135
Map<String, String> customLabels = Maps.newHashMap();
136+
customLabels.put(CLUSTER_LABEL_CLAIM, v1alpha1Function.getSpec().getClusterName());
133137
customLabels.put(TENANT_LABEL_CLAIM, tenant);
134138
customLabels.put(NAMESPACE_LABEL_CLAIM, namespace);
139+
customLabels.put(COMPONENT_LABEL_CLAIM, functionName);
135140
V1alpha1FunctionSpecPod pod = new V1alpha1FunctionSpecPod();
136141
if (worker().getFactoryConfig() != null && worker().getFactoryConfig().getCustomLabels() != null) {
137142
customLabels.putAll(worker().getFactoryConfig().getCustomLabels());
138143
}
139144
pod.setLabels(customLabels);
140145
v1alpha1Function.getSpec().setPod(pod);
146+
v1alpha1Function.getMetadata().setLabels(customLabels);
141147
try {
142148
this.upsertFunction(tenant, namespace, functionName, functionConfig, v1alpha1Function, clientAuthenticationDataHttps);
143149
Call call = worker().getCustomObjectsApi().createNamespacedCustomObjectCall(
@@ -174,32 +180,41 @@ public void updateFunction(final String tenant,
174180
validateUpdateFunctionRequestParams(tenant, namespace, functionName, functionConfig, uploadedInputStream != null);
175181

176182
try {
177-
Call getCall = worker().getCustomObjectsApi().getNamespacedCustomObjectCall(
178-
group,
179-
version,
180-
KubernetesUtils.getNamespace(worker().getFactoryConfig()),
181-
plural,
182-
functionName,
183-
null
184-
);
185-
V1alpha1Function oldFn = executeCall(getCall, V1alpha1Function.class);
183+
String cluster = worker().getWorkerConfig().getPulsarFunctionsCluster();
186184
V1alpha1Function v1alpha1Function = FunctionsUtil.createV1alpha1FunctionFromFunctionConfig(
187185
kind,
188186
group,
189187
version,
190188
functionName,
191189
functionPkgUrl,
192190
functionConfig,
193-
worker().getWorkerConfig().getFunctionsWorkerServiceCustomConfigs()
191+
worker().getWorkerConfig().getFunctionsWorkerServiceCustomConfigs(),
192+
cluster
194193
);
194+
Call getCall = worker().getCustomObjectsApi().getNamespacedCustomObjectCall(
195+
group,
196+
version,
197+
KubernetesUtils.getNamespace(worker().getFactoryConfig()),
198+
plural,
199+
v1alpha1Function.getMetadata().getName(),
200+
null
201+
);
202+
V1alpha1Function oldFn = executeCall(getCall, V1alpha1Function.class);
203+
204+
v1alpha1Function.getMetadata().setNamespace(KubernetesUtils.getNamespace(worker().getFactoryConfig()));
205+
Map<String, String> customLabels = oldFn.getMetadata().getLabels();
206+
V1alpha1FunctionSpecPod pod = new V1alpha1FunctionSpecPod();
207+
pod.setLabels(customLabels);
208+
v1alpha1Function.getSpec().setPod(pod);
209+
v1alpha1Function.getMetadata().setLabels(customLabels);
195210
v1alpha1Function.getMetadata().setResourceVersion(oldFn.getMetadata().getResourceVersion());
196211
this.upsertFunction(tenant, namespace, functionName, functionConfig, v1alpha1Function, clientAuthenticationDataHttps);
197212
Call replaceCall = worker().getCustomObjectsApi().replaceNamespacedCustomObjectCall(
198213
group,
199214
version,
200215
KubernetesUtils.getNamespace(worker().getFactoryConfig()),
201216
plural,
202-
functionName,
217+
v1alpha1Function.getMetadata().getName(),
203218
v1alpha1Function,
204219
null,
205220
null,
@@ -227,13 +242,14 @@ public FunctionConfig getFunctionInfo(final String tenant,
227242
clientAuthenticationDataHttps,
228243
ComponentTypeUtils.toString(componentType));
229244

245+
String hashName = CommonUtil.generateObjectName(worker(), tenant, namespace, componentName);
230246
try {
231247
Call call = worker().getCustomObjectsApi().getNamespacedCustomObjectCall(
232248
group,
233249
version,
234250
KubernetesUtils.getNamespace(worker().getFactoryConfig()),
235251
plural,
236-
componentName,
252+
hashName,
237253
null
238254
);
239255
V1alpha1Function v1alpha1Function = executeCall(call, V1alpha1Function.class);
@@ -270,10 +286,11 @@ public FunctionStatus getFunctionStatus(final String tenant,
270286
clientAuthenticationDataHttps,
271287
ComponentTypeUtils.toString(componentType));
272288
FunctionStatus functionStatus = new FunctionStatus();
289+
String hashName = CommonUtil.generateObjectName(worker(), tenant, namespace, componentName);
273290
try {
274291
Call call = worker().getCustomObjectsApi().getNamespacedCustomObjectCall(
275292
group, version, KubernetesUtils.getNamespace(worker().getFactoryConfig()),
276-
plural, componentName, null);
293+
plural, hashName, null);
277294
V1alpha1Function v1alpha1Function = executeCall(call, V1alpha1Function.class);
278295
FunctionStatus.FunctionInstanceStatus functionInstanceStatus = new FunctionStatus.FunctionInstanceStatus();
279296
FunctionStatus.FunctionInstanceStatus.FunctionInstanceStatusData functionInstanceStatusData =

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

Lines changed: 15 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -18,11 +18,13 @@
1818
*/
1919
package io.functionmesh.compute.rest.api;
2020

21+
import com.google.common.collect.Maps;
2122
import io.functionmesh.compute.functions.models.V1alpha1Function;
2223
import io.functionmesh.compute.functions.models.V1alpha1FunctionList;
2324
import io.functionmesh.compute.MeshWorkerService;
2425
import io.functionmesh.compute.sinks.models.V1alpha1Sink;
2526
import io.functionmesh.compute.sources.models.V1alpha1Source;
27+
import io.functionmesh.compute.util.CommonUtil;
2628
import io.functionmesh.compute.util.KubernetesUtils;
2729
import lombok.extern.slf4j.Slf4j;
2830
import okhttp3.Call;
@@ -49,6 +51,7 @@
4951
import java.net.URI;
5052
import java.util.LinkedList;
5153
import java.util.List;
54+
import java.util.Map;
5255
import java.util.concurrent.ExecutionException;
5356
import java.util.function.Supplier;
5457

@@ -109,28 +112,14 @@ public void deregisterFunction(final String tenant,
109112
clientAuthenticationDataHttps,
110113
ComponentTypeUtils.toString(componentType));
111114
try {
112-
Call componentCall =
113-
worker().getCustomObjectsApi()
114-
.getNamespacedCustomObjectCall(
115-
group, version, KubernetesUtils.getNamespace(worker().getFactoryConfig()),
116-
plural, componentName, null);
117-
String clusterName;
118-
if ("Source".equals(kind)) {
119-
V1alpha1Source v1alpha1Source = executeCall(componentCall, V1alpha1Source.class);
120-
clusterName = v1alpha1Source.getSpec().getClusterName();
121-
} else if ("Sink".equals(kind)) {
122-
V1alpha1Sink v1alpha1Sink = executeCall(componentCall, V1alpha1Sink.class);
123-
clusterName = v1alpha1Sink.getSpec().getClusterName();
124-
} else {
125-
V1alpha1Function v1alpha1Function = executeCall(componentCall, V1alpha1Function.class);
126-
clusterName = v1alpha1Function.getSpec().getClusterName();
127-
}
115+
String clusterName = worker().getWorkerConfig().getPulsarFunctionsCluster();
116+
String hashName = CommonUtil.createObjectName(clusterName, tenant, namespace, componentName);
128117
Call deleteObjectCall = worker().getCustomObjectsApi().deleteNamespacedCustomObjectCall(
129118
group,
130119
version,
131120
KubernetesUtils.getNamespace(worker().getFactoryConfig()),
132121
plural,
133-
componentName,
122+
hashName,
134123
null,
135124
null,
136125
null,
@@ -317,22 +306,29 @@ public List<String> listFunctions(final String tenant,
317306
final AuthenticationDataSource clientAuthenticationDataHttps) {
318307
List<String> result = new LinkedList<>();
319308
try {
309+
String labelSelector;
310+
String cluster = worker().getWorkerConfig().getPulsarFunctionsCluster();
311+
labelSelector = String.format(
312+
"%s=%s,%s=%s,%s=%s",
313+
CLUSTER_LABEL_CLAIM, cluster,
314+
TENANT_LABEL_CLAIM, tenant,
315+
NAMESPACE_LABEL_CLAIM, namespace);
320316
Call call = worker().getCustomObjectsApi().listNamespacedCustomObjectCall(
321317
group,
322318
version,
323319
KubernetesUtils.getNamespace(worker().getFactoryConfig()), plural,
324320
"false",
325321
null,
326322
null,
327-
null,
323+
labelSelector,
328324
null,
329325
null,
330326
null,
331327
false,
332328
null);
333329

334330
V1alpha1FunctionList list = executeCall(call, V1alpha1FunctionList.class);
335-
list.getItems().forEach(n -> result.add(n.getMetadata().getName()));
331+
list.getItems().forEach(n -> result.add(n.getMetadata().getLabels().get(COMPONENT_LABEL_CLAIM)));
336332
} catch (Exception e) {
337333
log.error("failed to fetch functions list from namespace {}, error message: {}", namespace, e.getMessage());
338334
}

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

Lines changed: 28 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -24,10 +24,12 @@
2424
import io.functionmesh.compute.sinks.models.V1alpha1SinkSpecPod;
2525
import io.functionmesh.compute.sinks.models.V1alpha1SinkSpecPodVolumeMounts;
2626
import io.functionmesh.compute.sinks.models.V1alpha1SinkSpecPodVolumes;
27+
import io.functionmesh.compute.util.CommonUtil;
2728
import io.functionmesh.compute.util.KubernetesUtils;
2829
import io.functionmesh.compute.util.SinksUtil;
2930
import io.functionmesh.compute.MeshWorkerService;
3031
import io.functionmesh.compute.sinks.models.V1alpha1SinkStatus;
32+
import io.kubernetes.client.openapi.apis.CustomObjectsApi;
3133
import lombok.extern.slf4j.Slf4j;
3234
import okhttp3.Call;
3335
import org.apache.commons.lang3.StringUtils;
@@ -123,6 +125,7 @@ public void registerSink(
123125
clientAuthenticationDataHttps,
124126
ComponentTypeUtils.toString(componentType));
125127
this.validateTenantIsExist(tenant, namespace, sinkName, clientRole);
128+
String cluster = worker().getWorkerConfig().getPulsarFunctionsCluster();
126129
V1alpha1Sink v1alpha1Sink =
127130
SinksUtil.createV1alpha1SkinFromSinkConfig(
128131
kind,
@@ -133,7 +136,7 @@ public void registerSink(
133136
uploadedInputStream,
134137
sinkConfig,
135138
this.meshWorkerServiceSupplier.get().getConnectorsManager(),
136-
worker().getWorkerConfig().getFunctionsWorkerServiceCustomConfigs());
139+
worker().getWorkerConfig().getFunctionsWorkerServiceCustomConfigs(), cluster);
137140
// override namesapce by configuration
138141
v1alpha1Sink.getMetadata().setNamespace(KubernetesUtils.getNamespace(worker().getFactoryConfig()));
139142
try {
@@ -147,6 +150,7 @@ public void registerSink(
147150
customLabels.putAll(worker().getFactoryConfig().getCustomLabels());
148151
}
149152
pod.setLabels(customLabels);
153+
v1alpha1Sink.getMetadata().setLabels(customLabels);
150154
v1alpha1Sink.getSpec().setPod(pod);
151155
this.upsertSink(tenant, namespace, sinkName, sinkConfig, v1alpha1Sink, clientAuthenticationDataHttps);
152156
Call call =
@@ -193,13 +197,8 @@ public void updateSink(
193197
clientRole,
194198
clientAuthenticationDataHttps,
195199
ComponentTypeUtils.toString(componentType));
200+
String cluster = worker().getWorkerConfig().getPulsarFunctionsCluster();
196201
try {
197-
Call getCall =
198-
worker().getCustomObjectsApi()
199-
.getNamespacedCustomObjectCall(
200-
group, version, namespace, plural, sinkName, null);
201-
V1alpha1Sink oldRes = executeCall(getCall, V1alpha1Sink.class);
202-
203202
V1alpha1Sink v1alpha1Sink =
204203
SinksUtil.createV1alpha1SkinFromSinkConfig(
205204
kind,
@@ -209,20 +208,33 @@ public void updateSink(
209208
sinkPkgUrl,
210209
uploadedInputStream,
211210
sinkConfig, this.meshWorkerServiceSupplier.get().getConnectorsManager(),
212-
worker().getWorkerConfig().getFunctionsWorkerServiceCustomConfigs());
211+
worker().getWorkerConfig().getFunctionsWorkerServiceCustomConfigs(), cluster);
212+
CustomObjectsApi customObjectsApi = worker().getCustomObjectsApi();
213+
Call getCall =
214+
customObjectsApi.getNamespacedCustomObjectCall(
215+
group, version,
216+
KubernetesUtils.getNamespace(worker().getFactoryConfig()), plural,
217+
v1alpha1Sink.getMetadata().getName(), null);
218+
V1alpha1Sink oldRes = executeCall(getCall, V1alpha1Sink.class);
219+
if (oldRes.getMetadata() == null || oldRes.getMetadata().getLabels() == null) {
220+
throw new RestException(Response.Status.NOT_FOUND, "This sink resource was not found");
221+
}
222+
Map<String, String> labels = oldRes.getMetadata().getLabels();
223+
V1alpha1SinkSpecPod pod = new V1alpha1SinkSpecPod();
224+
pod.setLabels(labels);
225+
v1alpha1Sink.getMetadata().setLabels(labels);
226+
v1alpha1Sink.getSpec().setPod(pod);
213227
this.upsertSink(tenant, namespace, sinkName, sinkConfig, v1alpha1Sink, clientAuthenticationDataHttps);
214228
v1alpha1Sink.getMetadata().setNamespace(KubernetesUtils.getNamespace(worker().getFactoryConfig()));
215229
v1alpha1Sink
216230
.getMetadata()
217231
.setResourceVersion(oldRes.getMetadata().getResourceVersion());
218-
Call replaceCall =
219-
worker().getCustomObjectsApi()
220-
.replaceNamespacedCustomObjectCall(
232+
Call replaceCall = customObjectsApi.replaceNamespacedCustomObjectCall(
221233
group,
222234
version,
223235
KubernetesUtils.getNamespace(worker().getFactoryConfig()),
224236
plural,
225-
sinkName,
237+
v1alpha1Sink.getMetadata().getName(),
226238
v1alpha1Sink,
227239
null,
228240
null,
@@ -270,11 +282,12 @@ public SinkStatus getSinkStatus(
270282
clientAuthenticationDataHttps,
271283
ComponentTypeUtils.toString(componentType));
272284
try {
285+
String hashName = CommonUtil.generateObjectName(worker(), tenant, namespace, componentName);
273286
Call call =
274287
worker().getCustomObjectsApi()
275288
.getNamespacedCustomObjectCall(
276289
group, version, KubernetesUtils.getNamespace(worker().getFactoryConfig()),
277-
plural, componentName, null);
290+
plural, hashName, null);
278291
V1alpha1Sink v1alpha1Sink = executeCall(call, V1alpha1Sink.class);
279292
SinkStatus.SinkInstanceStatus sinkInstanceStatus = new SinkStatus.SinkInstanceStatus();
280293
SinkStatus.SinkInstanceStatus.SinkInstanceStatusData sinkInstanceStatusData =
@@ -314,11 +327,12 @@ public SinkConfig getSinkInfo(
314327
validateSinkEnabled();
315328
this.validateGetInfoRequestParams(tenant, namespace, componentName, kind);
316329
try {
330+
String hashName = CommonUtil.generateObjectName(worker(), tenant, namespace, componentName);
317331
Call call =
318332
worker().getCustomObjectsApi()
319333
.getNamespacedCustomObjectCall(
320334
group, version, KubernetesUtils.getNamespace(worker().getFactoryConfig()),
321-
plural, componentName, null);
335+
plural, hashName, null);
322336

323337
V1alpha1Sink v1alpha1Sink = executeCall(call, V1alpha1Sink.class);
324338
return SinksUtil.createSinkConfigFromV1alpha1Sink(

0 commit comments

Comments
 (0)