Skip to content

Commit

Permalink
chore: option to disable node viewer in enterprise deployments (#14879)
Browse files Browse the repository at this point in the history
  • Loading branch information
@clnoll
clnoll committed Jan 3, 2025
1 parent 6b61cf1 commit 6f26202
Show file tree
Hide file tree
Showing 8 changed files with 363 additions and 22 deletions.
29 changes: 24 additions & 5 deletions ...server/src/main/java/io/airbyte/commons/server/handlers/InstanceConfigurationHandler.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,8 @@
import io.airbyte.commons.enums.Enums;
import io.airbyte.commons.license.ActiveAirbyteLicense;
import io.airbyte.commons.license.AirbyteLicense;
import io.airbyte.commons.server.helpers.KubernetesClientPermissionHelper;
import io.airbyte.commons.server.helpers.PermissionDeniedException;
import io.airbyte.commons.version.AirbyteVersion;
import io.airbyte.config.AuthenticatedUser;
import io.airbyte.config.Configs.AirbyteEdition;
Expand All @@ -32,7 +34,10 @@
import io.airbyte.data.exceptions.ConfigNotFoundException;
import io.airbyte.data.services.PermissionService;
import io.airbyte.validation.json.JsonValidationException;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.api.model.Node;
import io.fabric8.kubernetes.api.model.NodeList;
import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation;
import io.fabric8.kubernetes.client.dsl.Resource;
import io.micronaut.context.annotation.Value;
import jakarta.inject.Named;
import jakarta.inject.Singleton;
Expand Down Expand Up @@ -69,7 +74,7 @@ public class InstanceConfigurationHandler {
private final AuthConfigs authConfigs;
private final PermissionService permissionService;
private final Clock clock;
private final Optional<KubernetesClient> kubernetesClient;
private final Optional<KubernetesClientPermissionHelper> kubernetesClientPermissionHelper;

public InstanceConfigurationHandler(@Named("airbyteUrl") final Optional<String> airbyteUrl,
@Value("${airbyte.tracking.strategy:}") final String trackingStrategy,
Expand All @@ -83,7 +88,7 @@ public InstanceConfigurationHandler(@Named("airbyteUrl") final Optional<String>
final AuthConfigs authConfigs,
final PermissionService permissionService,
final Optional<Clock> clock,
final Optional<KubernetesClient> kubernetesClient) {
final Optional<KubernetesClientPermissionHelper> kubernetesClientPermissionHelper) {
this.airbyteUrl = airbyteUrl;
this.trackingStrategy = trackingStrategy;
this.airbyteEdition = airbyteEdition;
Expand All @@ -96,7 +101,7 @@ public InstanceConfigurationHandler(@Named("airbyteUrl") final Optional<String>
this.authConfigs = authConfigs;
this.permissionService = permissionService;
this.clock = clock.orElse(Clock.systemUTC());
this.kubernetesClient = kubernetesClient;
this.kubernetesClientPermissionHelper = kubernetesClientPermissionHelper;
}

private static final Logger LOGGER = LoggerFactory.getLogger(InstanceConfigurationHandler.class);
Expand Down Expand Up @@ -271,7 +276,21 @@ LicenseStatus currentLicenseStatus() {
}

private Integer nodesUsage() {
return kubernetesClient.map(client -> client.nodes().list().getItems().size()).orElse(null);
try {
final NonNamespaceOperation<Node, NodeList, Resource<Node>> nodes =
this.kubernetesClientPermissionHelper
.map(KubernetesClientPermissionHelper::listNodes)
.orElse(null);

if (nodes != null) {
return nodes.list().getItems().size();
}
} catch (PermissionDeniedException e) {
LOGGER.warn("Permission denied while attempting to get node usage: {}", e.getMessage());
} catch (Exception e) {
LOGGER.error("Unexpected error while fetching Kubernetes nodes: {}", e.getMessage(), e);
}
return null;
}

}
36 changes: 26 additions & 10 deletions ...ommons-server/src/main/kotlin/io/airbyte/commons/server/handlers/DiagnosticToolHandler.kt
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,8 @@ package io.airbyte.commons.server.handlers

import io.airbyte.api.model.generated.ActorType
import io.airbyte.commons.csp.CspChecker
import io.airbyte.commons.server.helpers.KubernetesClientPermissionHelper
import io.airbyte.commons.server.helpers.PermissionDeniedException
import io.airbyte.commons.yaml.Yamls
import io.airbyte.config.DestinationConnection
import io.airbyte.config.SourceConnection
Expand All @@ -16,8 +18,12 @@ import io.airbyte.data.services.ConnectionService
import io.airbyte.data.services.DestinationService
import io.airbyte.data.services.SourceService
import io.airbyte.data.services.WorkspaceService
import io.fabric8.kubernetes.api.model.Node
import io.fabric8.kubernetes.api.model.NodeList
import io.fabric8.kubernetes.api.model.Quantity
import io.fabric8.kubernetes.client.KubernetesClient
import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation
import io.fabric8.kubernetes.client.dsl.Resource
import jakarta.inject.Singleton
import org.yaml.snakeyaml.DumperOptions
import org.yaml.snakeyaml.Yaml
Expand Down Expand Up @@ -50,6 +56,7 @@ open class DiagnosticToolHandler(
private val actorDefinitionVersionHelper: ActorDefinitionVersionHelper,
private val instanceConfigurationHandler: InstanceConfigurationHandler,
private val kubernetesClient: KubernetesClient,
private val kubernetesClientPermissionHelper: KubernetesClientPermissionHelper,
private val cspChecker: CspChecker,
) {
private val yaml = Yaml(DumperOptions().apply { defaultFlowStyle = DumperOptions.FlowStyle.BLOCK })
Expand Down Expand Up @@ -84,7 +91,14 @@ open class DiagnosticToolHandler(
logger.error { "Error in writing airbyte instance yaml. Message: ${it.message}" }
}
runCatching {
addAirbyteDeploymentYaml(zipOut)
val nodes =
try {
kubernetesClientPermissionHelper.listNodes()
} catch (e: PermissionDeniedException) {
logger.warn { "Skipping writing deployment yaml; node viewer permission denied. Message: ${e.message}" }
null
}
nodes?.let { addAirbyteDeploymentYaml(zipOut, it) }
}.onFailure {
logger.error { "Error in writing deployment yaml. Message: ${it.message}" }
}
Expand Down Expand Up @@ -268,36 +282,38 @@ open class DiagnosticToolHandler(
return licenseInfo
}

private fun addAirbyteDeploymentYaml(zipOut: ZipOutputStream) {
private fun addAirbyteDeploymentYaml(
zipOut: ZipOutputStream,
nodes: NonNamespaceOperation<Node, NodeList, Resource<Node>>?,
) {
val zipEntry = ZipEntry(AIRBYTE_DEPLOYMENT_YAML)
zipOut.putNextEntry(zipEntry)
val deploymentYamlContent = generateDeploymentYaml()
val deploymentYamlContent = generateDeploymentYaml(nodes)
zipOut.write(deploymentYamlContent.toByteArray())
zipOut.closeEntry()
}

private fun generateDeploymentYaml(): String {
private fun generateDeploymentYaml(nodes: NonNamespaceOperation<Node, NodeList, Resource<Node>>?): String {
// Collect cluster information
val deploymentYamlData = mapOf("k8s" to collectK8sInfo())
val deploymentYamlData = mapOf("k8s" to collectK8sInfo(nodes))
return yaml.dump(deploymentYamlData)
}

private fun collectK8sInfo(): Map<String, Any> {
private fun collectK8sInfo(nodes: NonNamespaceOperation<Node, NodeList, Resource<Node>>?): Map<String, Any> {
logger.info { "Collecting k8s data..." }
val kubernetesInfo =
mapOf(
"nodes" to collectNodeInfo(kubernetesClient),
"nodes" to collectNodeInfo(nodes),
"pods" to collectPodInfo(kubernetesClient),
)
return kubernetesInfo
}

private fun collectNodeInfo(client: KubernetesClient): List<Map<String, Any>> {
private fun collectNodeInfo(nodes: NonNamespaceOperation<Node, NodeList, Resource<Node>>?): List<Map<String, Any>> {
logger.info { "Collecting nodes data..." }

val nodeList: List<Map<String, Any>> =
client
.nodes()
nodes
?.list()
?.items
?.map { node ->
Expand Down
42 changes: 42 additions & 0 deletions ...ver/src/main/kotlin/io/airbyte/commons/server/helpers/KubernetesClientPermissionHelper.kt
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,42 @@
package io.airbyte.commons.server.helpers

import io.fabric8.kubernetes.api.model.Node
import io.fabric8.kubernetes.api.model.NodeList
import io.fabric8.kubernetes.api.model.authorization.v1.SelfSubjectAccessReview
import io.fabric8.kubernetes.api.model.authorization.v1.SelfSubjectAccessReviewSpec
import io.fabric8.kubernetes.client.KubernetesClient
import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation
import io.fabric8.kubernetes.client.dsl.Resource
import jakarta.inject.Inject
import jakarta.inject.Singleton

class PermissionDeniedException(message: String) : RuntimeException(message)

@Singleton
class KubernetesClientPermissionHelper
@Inject
constructor(
private val kubernetesClient: KubernetesClient,
) {
fun listNodes(): NonNamespaceOperation<Node, NodeList, Resource<Node>>? {
if (!allowedToListNodes()) {
throw PermissionDeniedException("Permission denied: unable to list Kubernetes nodes.")
}
return kubernetesClient.nodes()
}

private fun allowedToListNodes(): Boolean {
val review = SelfSubjectAccessReview()
review.spec =
SelfSubjectAccessReviewSpec().apply {
resourceAttributes =
io.fabric8.kubernetes.api.model.authorization.v1.ResourceAttributes().apply {
verb = "list"
resource = "nodes"
}
}

val response = kubernetesClient.authorization().v1().selfSubjectAccessReview().create(review)
return response.status?.allowed ?: false
}
}
53 changes: 48 additions & 5 deletions ...er/src/test/java/io/airbyte/commons/server/handlers/InstanceConfigurationHandlerTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

Expand All @@ -24,6 +25,7 @@
import io.airbyte.commons.license.ActiveAirbyteLicense;
import io.airbyte.commons.license.AirbyteLicense;
import io.airbyte.commons.license.AirbyteLicense.LicenseType;
import io.airbyte.commons.server.helpers.KubernetesClientPermissionHelper;
import io.airbyte.commons.version.AirbyteVersion;
import io.airbyte.config.AuthenticatedUser;
import io.airbyte.config.Configs.AirbyteEdition;
Expand All @@ -37,11 +39,15 @@
import io.airbyte.data.exceptions.ConfigNotFoundException;
import io.airbyte.data.services.PermissionService;
import io.airbyte.validation.json.JsonValidationException;
import io.fabric8.kubernetes.api.model.Node;
import io.fabric8.kubernetes.api.model.NodeList;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation;
import java.io.IOException;
import java.time.Clock;
import java.time.Instant;
import java.time.ZoneId;
import java.util.Arrays;
import java.util.Date;
import java.util.Optional;
import java.util.UUID;
Expand Down Expand Up @@ -86,6 +92,10 @@ class InstanceConfigurationHandlerTest {
private PermissionService permissionService;
@Mock
private Optional<KubernetesClient> mKubernetesClient;
@Mock
private Optional<KubernetesClientPermissionHelper> mKubernetesClientHelper;
@Mock
private KubernetesClientPermissionHelper kubernetesClientPermissionHelperMock;

private AirbyteKeycloakConfiguration keycloakConfiguration;
private ActiveAirbyteLicense activeAirbyteLicense;
Expand Down Expand Up @@ -171,7 +181,7 @@ void testGetInstanceConfigurationTrackingStrategy(final String envValue, final T
mAuthConfigs,
permissionService,
Optional.empty(),
mKubernetesClient);
mKubernetesClientHelper);

final var result = handler.getInstanceConfiguration();

Expand Down Expand Up @@ -302,6 +312,39 @@ void testLicenseInfo() {
assertEquals(licenseInfoResponse.getExpirationDate(), EXPIRATION_DATE.toInstant().getEpochSecond());
assertEquals(licenseInfoResponse.getMaxEditors(), MAX_EDITORS);
assertEquals(licenseInfoResponse.getMaxNodes(), MAX_NODES);
assertEquals(licenseInfoResponse.getUsedNodes(), null);
}

@Test
void testLicenseInfoWithUsedNodes() {
var mockNodesOperation = mock(NonNamespaceOperation.class);
var nodeList = new NodeList();
nodeList.setItems(Arrays.asList(new Node(), new Node(), new Node(), new Node(), new Node()));

when(kubernetesClientPermissionHelperMock.listNodes()).thenReturn(mockNodesOperation);
when(mockNodesOperation.list()).thenReturn(nodeList);

final var handler = new InstanceConfigurationHandler(
Optional.of(AIRBYTE_URL),
"logging",
AirbyteEdition.PRO,
new AirbyteVersion("0.50.1"),
Optional.of(activeAirbyteLicense),
mWorkspacePersistence,
mWorkspacesHandler,
mUserPersistence,
mOrganizationPersistence,
mAuthConfigs,
permissionService,
Optional.empty(),
Optional.of(kubernetesClientPermissionHelperMock));

final var licenseInfoResponse = handler.licenseInfo();

assertEquals(licenseInfoResponse.getExpirationDate(), EXPIRATION_DATE.toInstant().getEpochSecond());
assertEquals(licenseInfoResponse.getMaxEditors(), MAX_EDITORS);
assertEquals(licenseInfoResponse.getMaxNodes(), MAX_NODES);
assertEquals(licenseInfoResponse.getUsedNodes(), nodeList.getItems().size());
}

@Test
Expand All @@ -321,7 +364,7 @@ void testInvalidLicenseTest() {
mAuthConfigs,
permissionService,
Optional.empty(),
mKubernetesClient);
mKubernetesClientHelper);
assertEquals(handler.currentLicenseStatus(), LicenseStatus.INVALID);
}

Expand All @@ -340,7 +383,7 @@ void testExpiredLicenseTest() {
mAuthConfigs,
permissionService,
Optional.of(Clock.fixed(Instant.MAX, ZoneId.systemDefault())),
mKubernetesClient);
mKubernetesClientHelper);
assertEquals(handler.currentLicenseStatus(), LicenseStatus.EXPIRED);
}

Expand All @@ -360,7 +403,7 @@ void testExceededEditorsLicenseTest() {
mAuthConfigs,
permissionService,
Optional.empty(),
mKubernetesClient);
mKubernetesClientHelper);
when(permissionService.listPermissions()).thenReturn(
Stream.generate(UUID::randomUUID)
.map(userId -> new Permission().withUserId(userId).withPermissionType(Permission.PermissionType.ORGANIZATION_EDITOR))
Expand Down Expand Up @@ -408,7 +451,7 @@ private InstanceConfigurationHandler getInstanceConfigurationHandler(final boole
mAuthConfigs,
permissionService,
Optional.empty(),
mKubernetesClient);
mKubernetesClientHelper);
}

}
Loading

0 comments on commit 6f26202

Please sign in to comment.