Repository: smallrye/smallrye-stork (java)

Issue:
Knative SD implementation doesn't get the correct service instance endpoint
From [this PR](https://github.com/quarkusio/quarkus/pull/30272)

The knative service discovery strategy is not working correctly. Given the quarkus-super-heroes project, when curl the rest-fights app on the /api/fights/randomfighters endpoint I would expect to see two random fighters, but I don’t. I see the fallback fighters
```
╰─ http https://rest-fights-eric-deandrea-dev.apps.sandbox-m2.ll9k.p1.openshiftapps.com/api/fights/randomfighters
HTTP/1.1 200 OK
cache-control: private
content-length: 314
content-type: application/json;charset=UTF-8
date: Mon, 06 Feb 2023 16:39:59 GMT
server: envoy
set-cookie: f876558f2b90f5b848dd878960dd0dc7=539d5cd9b01519008a395f7357457294; path=/; HttpOnly
x-envoy-upstream-service-time: 859

{
    "hero": {
        "level": 1,
        "name": "Fallback hero",
        "picture": "https://dummyimage.com/280x380/1e8fff/ffffff&text=Fallback+Hero",
        "powers": "Fallback hero powers"
    },
    "villain": {
        "level": 45,
        "name": "Fallback villain",
        "picture": "https://dummyimage.com/280x380/b22222/ffffff&text=Fallback+Villain",
        "powers": "Fallback villain powers"
    }
}
```
To reproduce: 
using https://developers.redhat.com/developer-sandbox
https://github.com/edeandrea/quarkus-super-heroes/tree/fix-stork
Steps to deploy:
Make sure you’re using Java 17!
1 - oc apply -f https://raw.githubusercontent.com/edeandrea/quarkus-super-heroes/fix-stork/rest-villains/deploy/k8s/java17-knative.yml
2 - oc apply -f https://raw.githubusercontent.com/edeandrea/quarkus-super-heroes/fix-stork/rest-heroes/deploy/k8s/java17-knative.yml
3 - Clone https://github.com/edeandrea/quarkus-super-heroes and switch to the fix-stork branch
4 - 
```
cd quarkus-super-heroes/rest-fights
./mvnw clean package -DskipTests \
  -Dquarkus.profile=knative-openshift-17 \
  -Dquarkus.container-image.registry=image-registry.openshift-image-registry.svc:5000 \
  -Dquarkus.container-image.group=$(oc project -q) \
  -Dquarkus.knative.labels.\"app.openshift.io/runtime\"=quarkus \
  -Dquarkus.knative.version=1.0 \
  -Dquarkus.kubernetes.deploy=true
```

Pull request description:
Remove schema for host and use 0 as default port in knative
Fixes #488

Patch:
diff --git a/service-discovery/knative/pom.xml b/service-discovery/knative/pom.xml
index 190792a1..d4df8ecb 100644
--- a/service-discovery/knative/pom.xml
+++ b/service-discovery/knative/pom.xml
@@ -49,7 +49,6 @@
         <dependency>
             <groupId>io.fabric8</groupId>
             <artifactId>knative-client</artifactId>
-            <!--            <scope>test</scope>-->
         </dependency>
         <dependency>
             <groupId>io.vertx</groupId>
diff --git a/service-discovery/knative/src/main/java/io/smallrye/stork/servicediscovery/knative/KnativeServiceDiscovery.java b/service-discovery/knative/src/main/java/io/smallrye/stork/servicediscovery/knative/KnativeServiceDiscovery.java
index 97ff7b71..7808b3e3 100644
--- a/service-discovery/knative/src/main/java/io/smallrye/stork/servicediscovery/knative/KnativeServiceDiscovery.java
+++ b/service-discovery/knative/src/main/java/io/smallrye/stork/servicediscovery/knative/KnativeServiceDiscovery.java
@@ -1,5 +1,6 @@
 package io.smallrye.stork.servicediscovery.knative;
 
+import java.net.URI;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -147,9 +148,21 @@ public class KnativeServiceDiscovery extends CachingServiceDiscovery {
                         : Collections.emptyMap());
 
                 Metadata<KnativeMetadataKey> knativeMetadata = Metadata.of(KnativeMetadataKey.class);
+                String host = knService.getStatus().getUrl();
+                try {
+                    URI uri = new URI(knService.getStatus().getUrl());
+                    if (uri != null && uri.getScheme() != null) {
+                        host = uri.getHost();
+                        if (host == null) { // invalid URI
+                            throw new IllegalArgumentException("Invalid URL used: '" + uri + "'");
+                        }
+                    }
+                } catch (Exception e) {
+                    LOGGER.error(e.getMessage() + " for service: " + application);
+                }
 
                 serviceInstances
-                        .add(new DefaultServiceInstance(ServiceInstanceIds.next(), knService.getStatus().getUrl(), 8080, secure,
+                        .add(new DefaultServiceInstance(ServiceInstanceIds.next(), host, -1, secure,
                                 labels,
                                 knativeMetadata
                                         .with(KnativeMetadataKey.META_KNATIVE_SERVICE_ID, knService.getFullResourceName())


Test patch:
diff --git a/service-discovery/knative/src/test/java/io/smallrye/stork/servicediscovery/knative/KnativeServiceDiscoveryRealClusterTest.java b/service-discovery/knative/src/test/java/io/smallrye/stork/servicediscovery/knative/KnativeServiceDiscoveryRealClusterTest.java
new file mode 100644
index 00000000..00a1b4e1
--- /dev/null
+++ b/service-discovery/knative/src/test/java/io/smallrye/stork/servicediscovery/knative/KnativeServiceDiscoveryRealClusterTest.java
@@ -0,0 +1,96 @@
+package io.smallrye.stork.servicediscovery.knative;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
+import static org.awaitility.Awaitility.await;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+
+import io.fabric8.kubernetes.client.Config;
+import io.smallrye.stork.Stork;
+import io.smallrye.stork.api.ServiceInstance;
+import io.smallrye.stork.test.StorkTestUtils;
+import io.smallrye.stork.test.TestConfigProvider;
+
+@Disabled
+public class KnativeServiceDiscoveryRealClusterTest {
+
+    String k8sMasterUrl;
+    String namespace;
+
+    @BeforeEach
+    void setUp() {
+        TestConfigProvider.clear();
+        System.setProperty(Config.KUBERNETES_TRUST_CERT_SYSTEM_PROPERTY, "true");
+    }
+
+    @Test
+    void shouldDiscoverHeroesKnative() {
+        String svc = "hero-service";
+
+        TestConfigProvider.addServiceConfig(svc, null, "knative",
+                null, Map.of("knative-host", "https://api.sandbox-m2.ll9k.p1.openshiftapps.com:6443", "knative-namespace",
+                        "amunozhe-dev", "application", "rest-heroes"));
+
+        Stork stork = StorkTestUtils.getNewStorkInstance();
+
+        AtomicReference<List<ServiceInstance>> instances = new AtomicReference<>();
+
+        io.smallrye.stork.api.Service service = stork.getService(svc);
+        service.getServiceDiscovery().getServiceInstances()
+                .onFailure().invoke(th -> fail("Failed to get service instances from the cluster", th))
+                .subscribe().with(instances::set);
+
+        await().atMost(Duration.ofSeconds(40))
+                .until(() -> instances.get() != null);
+
+        assertThat(instances.get()).hasSize(1);
+        ServiceInstance svcInstance = instances.get().get(0);
+
+        // This code needs the smallrye-mutiny-vertx-web-client dependency
+        //        WebClient client = WebClient.create(Vertx.vertx(), new WebClientOptions()
+        //                .setDefaultHost(svcInstance.getHost()).setDefaultPort(svcInstance.getPort()).setSsl(false).setTrustAll(false));
+        //
+        //        await().untilAsserted(() -> Assertions.assertEquals(200,
+        //                client.get("/api/heroes/random").sendAndAwait().statusCode()));
+
+    }
+
+    //    @Test
+    void shouldDiscoverNamespacedKnativeServicesWithApp() {
+        String svc = "my-service";
+
+        TestConfigProvider.addServiceConfig("my-service", null, "knative",
+                null, Map.of("knative-namespace", "default", "application", "helloworld-go"));
+
+        Stork stork = StorkTestUtils.getNewStorkInstance();
+
+        AtomicReference<List<ServiceInstance>> instances = new AtomicReference<>();
+
+        io.smallrye.stork.api.Service service = stork.getService(svc);
+        service.getServiceDiscovery().getServiceInstances()
+                .onFailure().invoke(th -> fail("Failed to get service instances from the cluster", th))
+                .subscribe().with(instances::set);
+
+        await().atMost(Duration.ofSeconds(5))
+                .until(() -> instances.get() != null);
+
+        assertThat(instances.get()).hasSize(1);
+        ServiceInstance svcInstance = instances.get().get(0);
+
+        //        WebClient client = WebClient.create(Vertx.vertx(), new WebClientOptions()
+        //                .setDefaultHost(svcInstance.getHost()).setSsl(false).setTrustAll(false));
+        //
+        //        await().untilAsserted(() -> Assertions.assertEquals(200,
+        //                client.get("").sendAndAwait().statusCode()));
+
+    }
+
+}
diff --git a/service-discovery/knative/src/test/java/io/smallrye/stork/servicediscovery/knative/KnativeServiceDiscoveryTest.java b/service-discovery/knative/src/test/java/io/smallrye/stork/servicediscovery/knative/KnativeServiceDiscoveryTest.java
index 1b1c34be..31606b48 100644
--- a/service-discovery/knative/src/test/java/io/smallrye/stork/servicediscovery/knative/KnativeServiceDiscoveryTest.java
+++ b/service-discovery/knative/src/test/java/io/smallrye/stork/servicediscovery/knative/KnativeServiceDiscoveryTest.java
@@ -57,7 +57,7 @@ public class KnativeServiceDiscoveryTest {
 
         String knSvcName = "my-knservice";
 
-        registerKnativeServices(knSvcName, "http://hello.test.127.0.0.1.sslip.io", null);
+        registerKnativeServices(knSvcName, "hello.test.127.0.0.1.sslip.io", null);
 
         AtomicReference<List<ServiceInstance>> instances = new AtomicReference<>();
 
@@ -70,8 +70,8 @@ public class KnativeServiceDiscoveryTest {
                 .until(() -> instances.get() != null);
 
         assertThat(instances.get()).hasSize(1);
-        assertThat(instances.get().get(0).getHost()).isEqualTo("http://hello.test.127.0.0.1.sslip.io");
-        assertThat(instances.get().get(0).getPort()).isEqualTo(8080);
+        assertThat(instances.get().get(0).getHost()).isEqualTo("hello.test.127.0.0.1.sslip.io");
+        assertThat(instances.get().get(0).getPort()).isEqualTo(-1);
         Map<String, String> labels = instances.get().get(0).getLabels();
         assertThat(labels).contains(entry("serving.knative.dev/creator", "kubernetes-admin"),
                 entry("serving.knative.dev/lastModifier", "kubernetes-admin"));
@@ -109,8 +109,8 @@ public class KnativeServiceDiscoveryTest {
 
         String knativeService = "my-knservice";
 
-        registerKnativeServices(knativeService, "http://hello.ns1.127.0.0.1.sslip.io", "ns1");
-        registerKnativeServices(knativeService, "http://hello.ns2.127.0.0.1.sslip.io", "ns2");
+        registerKnativeServices(knativeService, "hello.ns1.127.0.0.1.sslip.io", "ns1");
+        registerKnativeServices(knativeService, "hello.ns2.127.0.0.1.sslip.io", "ns2");
 
         AtomicReference<List<ServiceInstance>> instances = new AtomicReference<>();
 
@@ -124,7 +124,7 @@ public class KnativeServiceDiscoveryTest {
 
         assertThat(instances.get()).hasSize(2);
         assertThat(instances.get().stream().map(ServiceInstance::getHost))
-                .containsExactlyInAnyOrder("http://hello.ns1.127.0.0.1.sslip.io", "http://hello.ns2.127.0.0.1.sslip.io");
+                .containsExactlyInAnyOrder("hello.ns1.127.0.0.1.sslip.io", "hello.ns2.127.0.0.1.sslip.io");
     }
 
     @Test
@@ -135,7 +135,7 @@ public class KnativeServiceDiscoveryTest {
 
         String knativeService = "my-knservice";
 
-        registerKnativeServices(knativeService, "http://hello.test.127.0.0.1.sslip.io", null);
+        registerKnativeServices(knativeService, "hello.test.127.0.0.1.sslip.io", null);
 
         AtomicReference<List<ServiceInstance>> instances = new AtomicReference<>();
 
@@ -148,8 +148,8 @@ public class KnativeServiceDiscoveryTest {
                 .until(() -> instances.get() != null);
 
         assertThat(instances.get()).hasSize(1);
-        assertThat(instances.get().get(0).getHost()).isEqualTo("http://hello.test.127.0.0.1.sslip.io");
-        assertThat(instances.get().get(0).getPort()).isEqualTo(8080);
+        assertThat(instances.get().get(0).getHost()).isEqualTo("hello.test.127.0.0.1.sslip.io");
+        assertThat(instances.get().get(0).getPort()).isEqualTo(-1);
         Map<String, String> labels = instances.get().get(0).getLabels();
         assertThat(labels).contains(entry("serving.knative.dev/creator", "kubernetes-admin"),
                 entry("serving.knative.dev/lastModifier", "kubernetes-admin"));
@@ -165,7 +165,7 @@ public class KnativeServiceDiscoveryTest {
 
         String knSvcName = "my-knservice";
 
-        registerKnativeServices(knSvcName, "http://hello.test.127.0.0.1.sslip.io", null);
+        registerKnativeServices(knSvcName, "hello.test.127.0.0.1.sslip.io", null);
 
         AtomicReference<List<ServiceInstance>> instances = new AtomicReference<>();
 
@@ -178,8 +178,8 @@ public class KnativeServiceDiscoveryTest {
                 .until(() -> instances.get() != null);
 
         assertThat(instances.get()).hasSize(1);
-        assertThat(instances.get().get(0).getHost()).isEqualTo("http://hello.test.127.0.0.1.sslip.io");
-        assertThat(instances.get().get(0).getPort()).isEqualTo(8080);
+        assertThat(instances.get().get(0).getHost()).isEqualTo("hello.test.127.0.0.1.sslip.io");
+        assertThat(instances.get().get(0).getPort()).isEqualTo(-1);
         Map<String, String> labels = instances.get().get(0).getLabels();
         assertThat(labels).contains(entry("serving.knative.dev/creator", "kubernetes-admin"),
                 entry("serving.knative.dev/lastModifier", "kubernetes-admin"));
@@ -204,7 +204,7 @@ public class KnativeServiceDiscoveryTest {
 
         String knSvcName = "my-knservice";
 
-        registerKnativeServices(knSvcName, "http://hello.test.127.0.0.1.sslip.io", null);
+        registerKnativeServices(knSvcName, "hello.test.127.0.0.1.sslip.io", null);
 
         AtomicReference<List<ServiceInstance>> instances = new AtomicReference<>();
 
@@ -217,8 +217,8 @@ public class KnativeServiceDiscoveryTest {
                 .until(() -> instances.get() != null);
 
         assertThat(instances.get()).hasSize(1);
-        assertThat(instances.get().get(0).getHost()).isEqualTo("http://hello.test.127.0.0.1.sslip.io");
-        assertThat(instances.get().get(0).getPort()).isEqualTo(8080);
+        assertThat(instances.get().get(0).getHost()).isEqualTo("hello.test.127.0.0.1.sslip.io");
+        assertThat(instances.get().get(0).getPort()).isEqualTo(-1);
         Map<String, String> labels = instances.get().get(0).getLabels();
         assertThat(labels).contains(entry("serving.knative.dev/creator", "kubernetes-admin"),
                 entry("serving.knative.dev/lastModifier", "kubernetes-admin"));
@@ -252,7 +252,7 @@ public class KnativeServiceDiscoveryTest {
         server.expect().get().withPath("/apis/serving.knative.dev/v1/namespaces/test/services/my-knservice")
                 .andReply(200, r -> {
                     serverHit.incrementAndGet();
-                    return buildKnService(knSvcName, "http://hello.test.127.0.0.1.sslip.io", "test");
+                    return buildKnService(knSvcName, "hello.test.127.0.0.1.sslip.io", "test");
                 }).always();
 
         TestConfigProvider.addServiceConfig("my-knservice", null, "knative",
diff --git a/service-discovery/kubernetes/src/test/java/io/smallrye/stork/servicediscovery/kubernetes/KubernetesServiceDiscoveryRealClusterIT.java b/service-discovery/kubernetes/src/test/java/io/smallrye/stork/servicediscovery/kubernetes/KubernetesServiceDiscoveryRealClusterIT.java
index da805e67..e6efd5e3 100644
--- a/service-discovery/kubernetes/src/test/java/io/smallrye/stork/servicediscovery/kubernetes/KubernetesServiceDiscoveryRealClusterIT.java
+++ b/service-discovery/kubernetes/src/test/java/io/smallrye/stork/servicediscovery/kubernetes/KubernetesServiceDiscoveryRealClusterIT.java
@@ -25,6 +25,7 @@ import io.smallrye.stork.api.ServiceRegistrar;
 import io.smallrye.stork.test.StorkTestUtils;
 import io.smallrye.stork.test.TestConfigProvider;
 
+@Disabled
 public class KubernetesServiceDiscoveryRealClusterIT {
 
     @BeforeEach
@@ -34,7 +35,6 @@ public class KubernetesServiceDiscoveryRealClusterIT {
     }
 
     @Test
-    @Disabled
     void shouldGetServiceFromK8sDefaultNamespace() {
 
         TestConfigProvider.addServiceConfig("rest-service", null, "kubernetes",
@@ -60,7 +60,6 @@ public class KubernetesServiceDiscoveryRealClusterIT {
     }
 
     @Test
-    @Disabled
     void shouldGetServicesForDefaultNamespaceOnNonSpecified() {
         String serviceName = "pod1";
 
@@ -82,7 +81,6 @@ public class KubernetesServiceDiscoveryRealClusterIT {
     }
 
     @Test
-    @Disabled("doesn't work yet")
     void shouldRegisterServiceInstancesInDefaultNamespace() throws InterruptedException {
         TestConfigProvider.addServiceConfig("svc", null, "kubernetes",
                 null, Map.of("k8s-host", "https://127.0.0.1:41711/", "k8s-namespace", "stork"));
diff --git a/service-discovery/kubernetes/src/test/java/io/smallrye/stork/servicediscovery/kubernetes/KubernetesServiceDiscoveryTest.java b/service-discovery/kubernetes/src/test/java/io/smallrye/stork/servicediscovery/kubernetes/KubernetesServiceDiscoveryTest.java
index 66f8d524..802830b0 100644
--- a/service-discovery/kubernetes/src/test/java/io/smallrye/stork/servicediscovery/kubernetes/KubernetesServiceDiscoveryTest.java
+++ b/service-discovery/kubernetes/src/test/java/io/smallrye/stork/servicediscovery/kubernetes/KubernetesServiceDiscoveryTest.java
@@ -98,6 +98,44 @@ public class KubernetesServiceDiscoveryTest {
         assertThat(instances.get()).allSatisfy(si -> assertThat(si.isSecure()).isFalse());
     }
 
+    @Test
+    void shouldGetServiceFromK8sWithApplicationNameConfig() {
+        TestConfigProvider.addServiceConfig("svc", null, "kubernetes",
+                null, Map.of("k8s-host", k8sMasterUrl, "k8s-namespace", defaultNamespace, "application", "greetingApp"));
+        Stork stork = StorkTestUtils.getNewStorkInstance();
+
+        String serviceName = "svc";
+        String[] ips = { "10.96.96.231", "10.96.96.232", "10.96.96.233" };
+
+        registerKubernetesResources("greetingApp", defaultNamespace, ips);
+
+        AtomicReference<List<ServiceInstance>> instances = new AtomicReference<>();
+
+        Service service = stork.getService(serviceName);
+        service.getServiceDiscovery().getServiceInstances()
+                .onFailure().invoke(th -> fail("Failed to get service instances from Kubernetes", th))
+                .subscribe().with(instances::set);
+
+        await().atMost(Duration.ofSeconds(5))
+                .until(() -> instances.get() != null);
+
+        assertThat(instances.get()).hasSize(3);
+        assertThat(instances.get().stream().map(ServiceInstance::getPort)).allMatch(p -> p == 8080);
+        assertThat(instances.get().stream().map(ServiceInstance::getHost)).containsExactlyInAnyOrder("10.96.96.231",
+                "10.96.96.232", "10.96.96.233");
+        for (ServiceInstance serviceInstance : instances.get()) {
+            Map<String, String> labels = serviceInstance.getLabels();
+            assertThat(labels).contains(entry("app.kubernetes.io/name", "greetingApp"),
+                    entry("app.kubernetes.io/version", "1.0"),
+                    entry("ui", "ui-" + ipAsSuffix(serviceInstance.getHost())));
+        }
+        instances.get().stream().map(ServiceInstance::getMetadata).forEach(metadata -> {
+            Metadata<KubernetesMetadataKey> k8sMetadata = (Metadata<KubernetesMetadataKey>) metadata;
+            assertThat(k8sMetadata.getMetadata()).containsKey(META_K8S_SERVICE_ID);
+        });
+        assertThat(instances.get()).allSatisfy(si -> assertThat(si.isSecure()).isFalse());
+    }
+
     @Test
     void shouldGetServiceFromK8sDefaultNamespaceUsingProgrammaticAPI() {
         Stork stork = StorkTestUtils.getNewStorkInstance();
@@ -472,23 +510,23 @@ public class KubernetesServiceDiscoveryTest {
         return result;
     }
 
-    private Endpoints buildAndRegisterKubernetesService(String serviceName, String namespace, boolean register,
+    private Endpoints buildAndRegisterKubernetesService(String applicationName, String namespace, boolean register,
             String... ipAdresses) {
 
         Map<String, String> serviceLabels = new HashMap<>();
-        serviceLabels.put("app.kubernetes.io/name", serviceName);
+        serviceLabels.put("app.kubernetes.io/name", applicationName);
         serviceLabels.put("app.kubernetes.io/version", "1.0");
 
         List<EndpointAddress> endpointAddresses = Arrays.stream(ipAdresses)
                 .map(ipAddress -> {
                     ObjectReference targetRef = new ObjectReference(null, null, "Pod",
-                            serviceName + "-" + ipAsSuffix(ipAddress), namespace, null, UUID.randomUUID().toString());
+                            applicationName + "-" + ipAsSuffix(ipAddress), namespace, null, UUID.randomUUID().toString());
                     EndpointAddress endpointAddress = new EndpointAddressBuilder().withIp(ipAddress).withTargetRef(targetRef)
                             .build();
                     return endpointAddress;
                 }).collect(Collectors.toList());
         Endpoints endpoint = new EndpointsBuilder()
-                .withNewMetadata().withName(serviceName).withLabels(serviceLabels).endMetadata()
+                .withNewMetadata().withName(applicationName).withLabels(serviceLabels).endMetadata()
                 .addToSubsets(new EndpointSubsetBuilder().withAddresses(endpointAddresses)
                         .addToPorts(new EndpointPortBuilder().withPort(8080).withProtocol("TCP").build())
                         .build())