[INLONG-8188][Manager] Support querying audit information by sink id

inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/inner/client/AuditClient.java

@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.inlong.manager.client.api.inner.client;
+
+import org.apache.inlong.manager.client.api.ClientConfiguration;
+import org.apache.inlong.manager.client.api.service.AuditApi;
+import org.apache.inlong.manager.client.api.util.ClientUtils;
+import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
+import org.apache.inlong.manager.common.util.Preconditions;
+import org.apache.inlong.manager.pojo.audit.AuditRequest;
+import org.apache.inlong.manager.pojo.audit.AuditVO;
+import org.apache.inlong.manager.pojo.common.Response;
+
+import java.util.List;
+
+/**
+ * Client for {@link AuditApi}.
+ */
+public class AuditClient {
+
+    private final AuditApi auditApi;
+
+    public AuditClient(ClientConfiguration configuration) {
+        auditApi = ClientUtils.createRetrofit(configuration).create(AuditApi.class);
+    }
+
+    /**
+     * Query audit data for list by condition
+     *
+     * @param request The audit request of query condition
+     * @return The result of query
+     */
+    public List<AuditVO> list(AuditRequest request) {
+        Preconditions.expectNotNull(request, "request cannot be null");
+        Preconditions.expectNotBlank(request.getInlongGroupId(), ErrorCodeEnum.INVALID_PARAMETER,
+                "inlong group id cannot be empty");
+        Preconditions.expectNotBlank(request.getInlongStreamId(), ErrorCodeEnum.INVALID_PARAMETER,
+                "inlong stream id cannot be empty");
+        Response<List<AuditVO>> response = ClientUtils.executeHttpCall(auditApi.list(request));
+        ClientUtils.assertRespSuccess(response);
+        return response.getData();
+    }
+
+    /**
+     * Refresh the base item of audit cache.
+     *
+     * @return true if not exception, or false if it has exception
+     */
+    public Boolean refreshCache(AuditRequest request) {
+        Response<Boolean> response = ClientUtils.executeHttpCall(auditApi.refreshCache());
+        ClientUtils.assertRespSuccess(response);
+        return response.getData();
+    }
+}

inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/inner/client/ClientFactory.java

@@ -52,6 +52,7 @@ public class ClientFactory {
     private final WorkflowApproverClient workflowApproverClient;
     private final WorkflowEventClient workflowEventClient;
     private final InlongConsumeClient consumeClient;
+    private final AuditClient auditClient;
 
     public ClientFactory(ClientConfiguration configuration) {
         groupClient = new InlongGroupClient(configuration);
@@ -68,5 +69,6 @@ public ClientFactory(ClientConfiguration configuration) {
         workflowApproverClient = new WorkflowApproverClient(configuration);
         workflowEventClient = new WorkflowEventClient(configuration);
         consumeClient = new InlongConsumeClient(configuration);
+        auditClient = new AuditClient(configuration);
     }
 }

inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/AuditApi.java

@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.inlong.manager.client.api.service;
+
+import org.apache.inlong.manager.pojo.audit.AuditRequest;
+import org.apache.inlong.manager.pojo.audit.AuditVO;
+import org.apache.inlong.manager.pojo.common.Response;
+
+import retrofit2.Call;
+import retrofit2.http.Body;
+import retrofit2.http.POST;
+
+import java.util.List;
+
+public interface AuditApi {
+
+    @POST("audit/list")
+    Call<Response<List<AuditVO>>> list(@Body AuditRequest auditRequest);
+
+    @POST("audit/refreshCache")
+    Call<Response<Boolean>> refreshCache();
+
+}

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/audit/AuditRequest.java

@@ -47,6 +47,9 @@ public class AuditRequest {
     @ApiModelProperty(value = "audit id list", required = true)
     private List<String> auditIds;
 
+    @ApiModelProperty(value = "sink id")
+    private Integer sinkId;
+
     @ApiModelProperty(value = "query date, format by 'yyyy-MM-dd'", required = true, example = "2022-01-01")
     @NotBlank(message = "dt not be blank")
     private String dt;

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AuditServiceImpl.java

@@ -184,10 +184,14 @@ public List<AuditVO> listByCondition(AuditRequest request) throws Exception {
 
         // for now, we use the first sink type only.
         // this is temporary behavior before multiple sinks in one stream is fully supported.
-        List<StreamSinkEntity> sinkEntityList = sinkEntityMapper.selectByRelatedId(groupId, streamId);
         String sinkNodeType = null;
-        if (CollectionUtils.isNotEmpty(sinkEntityList)) {
+        Integer sinkId = request.getSinkId();
+        if (sinkId == null) {
+            List<StreamSinkEntity> sinkEntityList = sinkEntityMapper.selectByRelatedId(groupId, streamId);
             sinkNodeType = sinkEntityList.get(0).getSinkType();
+        } else {
+            StreamSinkEntity sinkEntity = sinkEntityMapper.selectByPrimaryKey(sinkId);
+            sinkNodeType = sinkEntity.getSinkType();
         }
 
         // properly overwrite audit ids by role and stream config