Rc/2.9.0 (#91)

* Repository code addition #165 (#80)

* added repository code in the GQL analysis filter

* added repository code in the GQL analysis filter

* pom version update

---------

Co-authored-by: UmmulkiramR <[email protected]>

* Send kafka msg to ingest #164 (#81)

* new feature code changes - Added a mutation gql api, added kafka configurations, modified gql schema.

* new feature code changes - Added a mutation gql api, added kafka configurations, modified gql schema.

* to make kafka profile specific

* Refactored code to rectify failing tests

* Added a kafka placeholder to avoid tests from failing

---------

Co-authored-by: UmmulkiramR <[email protected]>

* Change kafka version (#84)

* new feature code changes - Added a mutation gql api, added kafka configurations, modified gql schema.

* new feature code changes - Added a mutation gql api, added kafka configurations, modified gql schema.

* to make kafka profile specific

* Refactored code to rectify failing tests

* Added a kafka placeholder to avoid tests from failing

* changed topic name.

* changed kafka version and null issue handling

* Changed kafka version and refactored code

* changed kafka release version

---------

Co-authored-by: UmmulkiramR <[email protected]>

* Change kafka version (#85)

* new feature code changes - Added a mutation gql api, added kafka configurations, modified gql schema.

* new feature code changes - Added a mutation gql api, added kafka configurations, modified gql schema.

* to make kafka profile specific

* Refactored code to rectify failing tests

* Added a kafka placeholder to avoid tests from failing

* changed topic name.

* changed kafka version and null issue handling

* Changed kafka version and refactored code

* changed kafka release version

* changed kafka release version

---------

Co-authored-by: UmmulkiramR <[email protected]>

* Change kafka version (#86)

* new feature code changes - Added a mutation gql api, added kafka configurations, modified gql schema.

* new feature code changes - Added a mutation gql api, added kafka configurations, modified gql schema.

* to make kafka profile specific

* Refactored code to rectify failing tests

* Added a kafka placeholder to avoid tests from failing

* changed topic name.

* changed kafka version and null issue handling

* Changed kafka version and refactored code

* changed kafka release version

* changed kafka release version

* changed topic name

---------

Co-authored-by: UmmulkiramR <[email protected]>

* changed type of 'analysis' field in mutation response (#87)

* new feature code changes - Added a mutation gql api, added kafka configurations, modified gql schema.

* new feature code changes - Added a mutation gql api, added kafka configurations, modified gql schema.

* to make kafka profile specific

* Refactored code to rectify failing tests

* Added a kafka placeholder to avoid tests from failing

* changed topic name.

* changed kafka version and null issue handling

* Changed kafka version and refactored code

* changed kafka release version

* changed kafka release version

* changed topic name

* changed type of 'analysis'  field in mutation response

---------

Co-authored-by: UmmulkiramR <[email protected]>

* release 2.9.0

* changed the port value

---------

Co-authored-by: UmmulkiramR <[email protected]>

pom.xml

@@ -10,7 +10,7 @@
     </parent>
     <groupId>bio.overture</groupId>
     <artifactId>song-search</artifactId>
-    <version>2.8.0</version>
+    <version>2.9.0</version>
     <name>song-search</name>
     <description>GQL microservice for searching maestro generated song indexes</description>
 
@@ -73,6 +73,13 @@
             <scope>test</scope>
         </dependency>
 
+        <!-- Kafka -->
+        <dependency>
+            <groupId>org.springframework.kafka</groupId>
+            <artifactId>spring-kafka</artifactId>
+            <version>2.8.4</version>
+        </dependency>
+
         <!-- Graphql java dependencies -->
         <dependency>
             <groupId>com.graphql-java</groupId>

src/main/java/bio/overture/songsearch/config/constants/SearchFields.java

@@ -26,6 +26,8 @@
 public class SearchFields {
   public static final String ANALYSIS_ID = "analysisId";
   public static final String ANALYSIS_TYPE = "analysisType";
+
+  public static final String REPOSITORY_CODE = "code";
   public static final String ANALYSIS_VERSION = "analysisVersion";
   public static final String ANALYSIS_STATE = "analysisState";
   public static final String FILE_OBJECT_ID = "objectId";

src/main/java/bio/overture/songsearch/config/kafka/AnalysisMessage.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright (c) 2018. Ontario Institute for Cancer Research
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package bio.overture.songsearch.config.kafka;
+
+import static lombok.AccessLevel.PRIVATE;
+
+import bio.overture.songsearch.model.enums.AnalysisState;
+import lombok.*;
+
+@Value
+// Note: although the AllArgs and NoArgs combination below seems odd,
+// it allows Jackson to deserialize to an immutable object without using any additional annotations.
+@AllArgsConstructor
+@NoArgsConstructor(force = true, access = PRIVATE)
+public class AnalysisMessage {
+
+  @NonNull private final String analysisId;
+  @NonNull private final String studyId;
+  @NonNull private final String state;
+  @NonNull private final String songServerId;
+  @NonNull private final Analysis analysis;
+
+  public static AnalysisMessage createAnalysisMessage(
+      bio.overture.songsearch.model.Analysis analysis, String songServerId) {
+
+    return new AnalysisMessage(
+        analysis.getAnalysisId(),
+        analysis.getStudyId(),
+        analysis.getAnalysisState().toString(),
+        analysis.getRepositories().get(0).getCode(),
+        new Analysis(
+            analysis.getAnalysisId(),
+            new AnalysisType(analysis.getAnalysisType()),
+            analysis.getAnalysisState(),
+            analysis.getStudyId()));
+  }
+
+  @Value
+  @AllArgsConstructor
+  private static class Analysis {
+    String analysisId;
+    AnalysisType analysisType;
+    AnalysisState analysisState;
+    String studyId;
+  }
+
+  @Data
+  @AllArgsConstructor
+  public static class AnalysisType {
+    String name;
+  }
+}

src/main/java/bio/overture/songsearch/config/kafka/DefaultSender.java

@@ -0,0 +1,13 @@
+package bio.overture.songsearch.config.kafka;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+public class DefaultSender implements Sender {
+
+  public void send(String payload, String key) {
+    log.debug("key: " + key);
+  }
+}

src/main/java/bio/overture/songsearch/config/kafka/KafkaConfig.java

@@ -0,0 +1,52 @@
+package bio.overture.songsearch.config.kafka;
+
+import java.util.HashMap;
+import java.util.Map;
+import lombok.val;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Profile;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+
+@Configuration
+@Profile("kafka")
+public class KafkaConfig {
+
+  @Value("${spring.kafka.bootstrap-servers}")
+  private String bootstrapServers;
+
+  @Value("${spring.kafka.template.automation-trigger}")
+  private String automation_trigger;
+
+  @Bean
+  public Map<String, Object> producerConfigs() {
+    val props = new HashMap<String, Object>();
+    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+
+    return props;
+  }
+
+  @Bean
+  public ProducerFactory<String, String> producerFactory() {
+    return new DefaultKafkaProducerFactory<>(producerConfigs());
+  }
+
+  @Bean
+  public KafkaTemplate<String, String> kafkaTemplate() {
+    val template = new KafkaTemplate<>(producerFactory());
+    template.setDefaultTopic(automation_trigger);
+    return template;
+  }
+
+  @Bean
+  public KafkaSender sender() {
+    return new KafkaSender();
+  }
+}

src/main/java/bio/overture/songsearch/config/kafka/KafkaSender.java

@@ -0,0 +1,17 @@
+package bio.overture.songsearch.config.kafka;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Profile;
+import org.springframework.kafka.core.KafkaTemplate;
+
+@Slf4j
+@Profile("kafka")
+public class KafkaSender implements Sender {
+  @Autowired private KafkaTemplate<String, String> kafkaTemplate;
+
+  public void send(String payload, String key) {
+    log.debug("sending payload='{}' to topic='{}'", payload, kafkaTemplate.getDefaultTopic());
+    kafkaTemplate.send(kafkaTemplate.getDefaultTopic(), key, payload);
+  }
+}

src/main/java/bio/overture/songsearch/config/kafka/Sender.java

@@ -0,0 +1,5 @@
+package bio.overture.songsearch.config.kafka;
+
+public interface Sender {
+  void send(String payload, String key);
+}

src/main/java/bio/overture/songsearch/graphql/GraphQLProvider.java

@@ -52,6 +52,8 @@ public class GraphQLProvider {
   private final AnalysisDataFetcher analysisDataFetcher;
   private final FileDataFetcher fileDataFetcher;
   private final EntityDataFetcher entityDataFetcher;
+
+  private final StartAutomationMutation startAutomationMutation;
   private final AuthProperties authProperties;
   private GraphQL graphQL;
   private GraphQLSchema graphQLSchema;
@@ -61,10 +63,12 @@ public GraphQLProvider(
       AnalysisDataFetcher analysisDataFetcher,
       FileDataFetcher fileDataFetcher,
       EntityDataFetcher entityDataFetcher,
+      StartAutomationMutation startAutomationMutation,
       AuthProperties authProperties) {
     this.analysisDataFetcher = analysisDataFetcher;
     this.fileDataFetcher = fileDataFetcher;
     this.entityDataFetcher = entityDataFetcher;
+    this.startAutomationMutation = startAutomationMutation;
     this.authProperties = authProperties;
   }
 
@@ -147,6 +151,9 @@ private RuntimeWiring buildWiring() {
                 .dataFetcher(
                     "sampleMatchedAnalysesForDonor",
                     analysisDataFetcher.getSampleMatchedAnalysesForDonorFetcher()))
+        .type(
+            newTypeWiring("Mutation")
+                .dataFetcher("startAutomation", startAutomationMutation.startAutomationResolver()))
         .build();
   }
 

src/main/java/bio/overture/songsearch/graphql/StartAutomationMutation.java

@@ -0,0 +1,108 @@
+/*
+ * Copyright (c) 2021 The Ontario Institute for Cancer Research. All rights reserved
+ *
+ * This program and the accompanying materials are made available under the terms of the GNU Affero General Public License v3.0.
+ * You should have received a copy of the GNU Affero General Public License along with
+ * this program. If not, see <http://www.gnu.org/licenses/>.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY
+ * EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
+ * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT
+ * SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
+ * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED
+ * TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
+ * OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER
+ * IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
+ * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package bio.overture.songsearch.graphql;
+
+import static bio.overture.songsearch.config.kafka.AnalysisMessage.createAnalysisMessage;
+import static bio.overture.songsearch.utils.JacksonUtils.convertValue;
+import static java.util.stream.Collectors.toUnmodifiableList;
+
+import bio.overture.songsearch.config.kafka.Sender;
+import bio.overture.songsearch.model.Analysis;
+import bio.overture.songsearch.model.AutomationMutationResponse;
+
+import bio.overture.songsearch.model.Sort;
+import bio.overture.songsearch.service.AnalysisService;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import graphql.schema.DataFetcher;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import lombok.NonNull;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import lombok.val;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+@Component
+@Slf4j
+public class StartAutomationMutation {
+
+  private final AnalysisService analysisService;
+
+  @Value("${songServerId")
+  private String songServerId;
+
+  private final Sender sender;
+
+  @Autowired
+  public StartAutomationMutation(AnalysisService analysisService, @NonNull Sender sender) {
+    this.analysisService = analysisService;
+    this.sender = sender;
+  }
+
+  public DataFetcher<AutomationMutationResponse> startAutomationResolver() {
+    return env -> {
+      val args = env.getArguments();
+
+      val filter = ImmutableMap.<String, Object>builder();
+      val page = ImmutableMap.<String, Integer>builder();
+      val sorts = ImmutableList.<Sort>builder();
+
+      if (args != null) {
+        if (args.get("filter") != null) filter.putAll((Map<String, Object>) args.get("filter"));
+        if (args.get("page") != null) page.putAll((Map<String, Integer>) args.get("page"));
+        if (args.get("sorts") != null) {
+          val rawSorts = (List<Object>) args.get("sorts");
+          sorts.addAll(
+              rawSorts.stream()
+                  .map(sort -> convertValue(sort, Sort.class))
+                  .collect(toUnmodifiableList()));
+        }
+      }
+
+      Analysis analysis =
+          analysisService.getAnalysisById(env.getArguments().get("analysisId").toString());
+      log.debug("Analysis fetched: " + analysis);
+
+      if (Objects.isNull(analysis)) {
+        log.debug("Analysis not found.");
+        AutomationMutationResponse response =
+            new AutomationMutationResponse(analysis, "Analysis not found");
+        return response;
+      }
+      AutomationMutationResponse response =
+          new AutomationMutationResponse(analysis, "payload sent");
+      sendAnalysisMessage(analysis);
+      log.debug("Message sent to kafka queue");
+      return response;
+    };
+  }
+
+  @SneakyThrows
+  public void sendAnalysisMessage(Analysis analysis) {
+    val message = createAnalysisMessage(analysis, songServerId);
+    log.debug("Message payload:: " + new ObjectMapper().writeValueAsString(message));
+    sender.send(new ObjectMapper().writeValueAsString(message), message.getAnalysisId());
+  }
+}

src/main/java/bio/overture/songsearch/model/AutomationMutationResponse.java

@@ -0,0 +1,10 @@
+package bio.overture.songsearch.model;
+
+import lombok.Value;
+
+@Value
+public class AutomationMutationResponse {
+
+  private Analysis analysis;
+  private String message;
+}

src/main/java/bio/overture/songsearch/repository/AnalysisRepository.java

@@ -80,6 +80,13 @@ private static Map<String, Function<Object, AbstractQueryBuilder<?>>> argumentPa
         .put(ANALYSIS_STATE, value -> new TermQueryBuilder("analysis_state", value))
         .put(STUDY_ID, value -> new TermQueryBuilder("study_id", value))
         .put(RUN_ID, value -> new TermQueryBuilder("workflow.run_id", value))
+        .put(
+            REPOSITORY_CODE,
+            value ->
+                new NestedQueryBuilder(
+                    "repositories",
+                    new TermQueryBuilder("repositories.code", value),
+                    ScoreMode.None))
         .put(
             DONOR_ID,
             value ->

src/main/java/bio/overture/songsearch/service/AnalysisService.java

@@ -35,19 +35,22 @@
 import lombok.NonNull;
 import lombok.SneakyThrows;
 import lombok.Value;
+import lombok.extern.slf4j.Slf4j;
 import lombok.val;
 import org.elasticsearch.action.search.MultiSearchResponse;
 import org.elasticsearch.search.SearchHit;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
 @Service
+@Slf4j
 public class AnalysisService {
 
   private final AnalysisRepository analysisRepository;
 
   @Autowired
   public AnalysisService(AnalysisRepository analysisRepository) {
+
     this.analysisRepository = analysisRepository;
   }