diff --git a/common/config/src/main/resources/kubernetes.conf b/common/config/src/main/resources/kubernetes.conf index e85924e570c..b45c6ffbfeb 100644 --- a/common/config/src/main/resources/kubernetes.conf +++ b/common/config/src/main/resources/kubernetes.conf @@ -41,6 +41,13 @@ kubernetes { max-num-of-running-computing-units-per-user = 10 max-num-of-running-computing-units-per-user = ${?MAX_NUM_OF_RUNNING_COMPUTING_UNITS_PER_USER} + # Terminate Kubernetes CUs whose latest workflow execution is older than this. + computing-unit-idle-timeout-minutes = 1440 + computing-unit-idle-timeout-minutes = ${?KUBERNETES_COMPUTING_UNIT_IDLE_TIMEOUT_MINUTES} + + computing-unit-idle-check-interval-minutes = 60 + computing-unit-idle-check-interval-minutes = ${?KUBERNETES_COMPUTING_UNIT_IDLE_CHECK_INTERVAL_MINUTES} + computing-unit-cpu-limit-options = "1,2,4" computing-unit-cpu-limit-options = ${?KUBERNETES_COMPUTING_UNIT_CPU_LIMIT_OPTIONS} diff --git a/common/config/src/main/scala/org/apache/texera/common/config/KubernetesConfig.scala b/common/config/src/main/scala/org/apache/texera/common/config/KubernetesConfig.scala index f6294767365..65203666399 100644 --- a/common/config/src/main/scala/org/apache/texera/common/config/KubernetesConfig.scala +++ b/common/config/src/main/scala/org/apache/texera/common/config/KubernetesConfig.scala @@ -38,6 +38,12 @@ object KubernetesConfig { val maxNumOfRunningComputingUnitsPerUser: Int = conf.getInt("kubernetes.max-num-of-running-computing-units-per-user") + val computingUnitIdleTimeoutMinutes: Long = + conf.getLong("kubernetes.computing-unit-idle-timeout-minutes") + + val computingUnitIdleCheckIntervalMinutes: Long = + conf.getLong("kubernetes.computing-unit-idle-check-interval-minutes") + val cpuLimitOptions: List[String] = conf .getString("kubernetes.computing-unit-cpu-limit-options") diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/ComputingUnitManagingService.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/ComputingUnitManagingService.scala index f0dffc89e11..ab6c2d84bcd 100644 --- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/ComputingUnitManagingService.scala +++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/ComputingUnitManagingService.scala @@ -23,7 +23,7 @@ import com.fasterxml.jackson.module.scala.DefaultScalaModule import io.dropwizard.configuration.{EnvironmentVariableSubstitutor, SubstitutingSourceProvider} import io.dropwizard.core.Application import io.dropwizard.core.setup.{Bootstrap, Environment} -import org.apache.texera.common.config.StorageConfig +import org.apache.texera.common.config.{KubernetesConfig, StorageConfig} import org.apache.texera.auth.{AuthFeatures, RequestLoggingFilter, RoleAnnotationEnforcer} import org.apache.texera.dao.SqlServer import org.apache.texera.service.resource.{ @@ -32,9 +32,44 @@ import org.apache.texera.service.resource.{ ComputingUnitManagingResource, HealthCheckResource } +import org.apache.texera.service.resource.ComputingUnitManagingResource.TerminatedComputingUnitInfo +import org.slf4j.LoggerFactory import java.nio.file.Path +import java.util.concurrent.TimeUnit class ComputingUnitManagingService extends Application[ComputingUnitManagingServiceConfiguration] { + private val logger = LoggerFactory.getLogger(classOf[ComputingUnitManagingService]) + + private[service] def initSqlServer( + connect: (String, String, String) => Unit = SqlServer.initConnection + ): Unit = + connect( + StorageConfig.jdbcUrl, + StorageConfig.jdbcUsername, + StorageConfig.jdbcPassword + ) + + private[service] def registerIdleComputingUnitCleanup(environment: Environment): Unit = + registerIdleComputingUnitCleanup((command, initialDelay, delay, unit) => + environment.lifecycle + .scheduledExecutorService("idle-computing-unit-terminator") + .threads(1) + .build() + .scheduleWithFixedDelay(command, initialDelay, delay, unit) + ) + + private[service] def registerIdleComputingUnitCleanup( + scheduleWithFixedDelay: (Runnable, Long, Long, TimeUnit) => Unit + ): Unit = + ComputingUnitManagingService.registerIdleComputingUnitCleanup( + scheduleWithFixedDelay, + KubernetesConfig.kubernetesComputingUnitEnabled, + KubernetesConfig.computingUnitIdleTimeoutMinutes, + KubernetesConfig.computingUnitIdleCheckIntervalMinutes, + () => ComputingUnitManagingResource.terminateIdleKubernetesComputingUnits(), + message => logger.info(message), + throwable => logger.warn("Failed to terminate idle Kubernetes computing units", throwable) + ) override def initialize( bootstrap: Bootstrap[ComputingUnitManagingServiceConfiguration] @@ -59,11 +94,7 @@ class ComputingUnitManagingService extends Application[ComputingUnitManagingServ AuthFeatures.register(environment) - SqlServer.initConnection( - StorageConfig.jdbcUrl, - StorageConfig.jdbcUsername, - StorageConfig.jdbcPassword - ) + initSqlServer() environment.jersey().register(new ComputingUnitManagingResource) environment.jersey().register(new ComputingUnitAccessResource) @@ -74,12 +105,83 @@ class ComputingUnitManagingService extends Application[ComputingUnitManagingServ "ComputingUnitManagingService" ) + registerIdleComputingUnitCleanup(environment) + // Route request logs through SLF4J, controlled by TEXERA_SERVICE_LOG_LEVEL RequestLoggingFilter.register(environment.getApplicationContext) } } object ComputingUnitManagingService { + private[service] def shouldScheduleIdleComputingUnitCleanup( + kubernetesComputingUnitEnabled: Boolean, + idleTimeoutMinutes: Long + ): Boolean = + kubernetesComputingUnitEnabled && idleTimeoutMinutes > 0 + + private[service] def registerIdleComputingUnitCleanup( + scheduleWithFixedDelay: (Runnable, Long, Long, TimeUnit) => Unit, + kubernetesComputingUnitEnabled: Boolean, + idleTimeoutMinutes: Long, + idleCheckIntervalMinutes: Long, + terminateIdleComputingUnits: () => List[TerminatedComputingUnitInfo], + logTerminatedUnits: String => Unit, + logCleanupFailure: Throwable => Unit + ): Unit = + if ( + shouldScheduleIdleComputingUnitCleanup(kubernetesComputingUnitEnabled, idleTimeoutMinutes) + ) { + scheduleIdleComputingUnitCleanup( + scheduleWithFixedDelay, + idleCheckIntervalMinutes, + terminateIdleComputingUnits, + logTerminatedUnits, + logCleanupFailure + ) + } + + private[service] def scheduleIdleComputingUnitCleanup( + scheduleWithFixedDelay: (Runnable, Long, Long, TimeUnit) => Unit, + idleCheckIntervalMinutes: Long, + terminateIdleComputingUnits: () => List[TerminatedComputingUnitInfo], + logTerminatedUnits: String => Unit, + logCleanupFailure: Throwable => Unit + ): Unit = + scheduleWithFixedDelay( + () => + runIdleComputingUnitCleanup( + terminateIdleComputingUnits, + logTerminatedUnits, + logCleanupFailure + ), + idleCheckIntervalMinutes, + idleCheckIntervalMinutes, + TimeUnit.MINUTES + ) + + private[service] def runIdleComputingUnitCleanup( + terminateIdleComputingUnits: () => List[TerminatedComputingUnitInfo], + logTerminatedUnits: String => Unit, + logCleanupFailure: Throwable => Unit + ): Unit = + try { + val terminated = terminateIdleComputingUnits() + if (terminated.nonEmpty) { + val terminatedDetails = terminated + .map(unit => + s"cuid=${unit.cuid}, name=${unit.name}, uid=${unit.uid}, username=${unit.username + .getOrElse("unknown")}, reason=${unit.reason.getLiteral}" + ) + .mkString("; ") + logTerminatedUnits( + s"Terminated ${terminated.size} idle Kubernetes computing unit(s): $terminatedDetails" + ) + } + } catch { + case t: Throwable => + logCleanupFailure(t) + } + def main(args: Array[String]): Unit = { val configFilePath = Path .of(sys.env.getOrElse("TEXERA_HOME", ".")) diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala index 3a249d296e0..9b06c2126fe 100644 --- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala +++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala @@ -42,9 +42,11 @@ import org.apache.texera.common.config.{ } import org.apache.texera.dao.SqlServer import org.apache.texera.dao.SqlServer.withTransaction +import org.apache.texera.dao.jooq.generated.Tables.{WORKFLOW_COMPUTING_UNIT, WORKFLOW_EXECUTIONS} import org.apache.texera.dao.jooq.generated.enums.{ PrivilegeEnum, UserRoleEnum, + WorkflowComputingUnitTerminationReasonEnum, WorkflowComputingUnitTypeEnum } import org.apache.texera.dao.jooq.generated.tables.daos.{ @@ -53,6 +55,7 @@ import org.apache.texera.dao.jooq.generated.tables.daos.{ WorkflowComputingUnitDao } import org.apache.texera.dao.jooq.generated.tables.pojos.WorkflowComputingUnit +import org.apache.texera.service.ComputingUnitManagingService import org.apache.texera.service.resource.ComputingUnitManagingResource._ import org.apache.texera.service.util.{ ComputingUnitHelpers, @@ -61,6 +64,8 @@ import org.apache.texera.service.util.{ KubernetesClient } import org.jooq.{DSLContext, EnumType} +import org.jooq.impl.DSL.max +import org.slf4j.LoggerFactory import play.api.libs.json._ import java.sql.Timestamp @@ -73,6 +78,136 @@ object ComputingUnitManagingResource { .getInstance() .createDSLContext() + private[resource] final class IdleComputingUnitCleanupConfig( + val enabled: Boolean, + val idleTimeoutMinutes: Long + ) { + def copy( + enabled: Boolean = this.enabled, + idleTimeoutMinutes: Long = this.idleTimeoutMinutes + ): IdleComputingUnitCleanupConfig = + new IdleComputingUnitCleanupConfig(enabled, idleTimeoutMinutes) + } + + private[resource] trait KubernetesPodOperations { + val podExists: Int => Boolean + val deletePod: Int => Unit + } + + private[resource] object DefaultKubernetesPodOperations extends KubernetesPodOperations { + private[resource] var podExistsDelegate: Int => Boolean = KubernetesClient.podExists + private[resource] var deletePodDelegate: Int => Unit = KubernetesClient.deletePod + + override val podExists: Int => Boolean = cuid => podExistsDelegate(cuid) + override val deletePod: Int => Unit = cuid => deletePodDelegate(cuid) + } + + private[resource] def lastComputingUnitActivityTime( + unit: WorkflowComputingUnit, + latestUpdateTime: Option[Timestamp], + latestStartTime: Option[Timestamp] + ): Timestamp = + Seq( + latestUpdateTime, + latestStartTime, + Option(unit.getCreationTime) + ).flatten.maxBy(_.getTime) + + private[resource] def shouldTerminateIdleComputingUnit( + hasActiveExecution: Boolean, + lastExecutionTime: Timestamp, + cutoff: Timestamp + ): Boolean = + !hasActiveExecution && lastExecutionTime.before(cutoff) + + def terminateIdleKubernetesComputingUnits(): List[TerminatedComputingUnitInfo] = + terminateIdleKubernetesComputingUnits( + new IdleComputingUnitCleanupConfig( + KubernetesConfig.kubernetesComputingUnitEnabled, + KubernetesConfig.computingUnitIdleTimeoutMinutes + ), + () => new Timestamp(System.currentTimeMillis()), + DefaultKubernetesPodOperations + ) + + private[resource] def terminateIdleKubernetesComputingUnits( + cleanupConfig: IdleComputingUnitCleanupConfig, + currentTime: () => Timestamp, + podOperations: KubernetesPodOperations + ): List[TerminatedComputingUnitInfo] = { + if (!cleanupConfig.enabled || cleanupConfig.idleTimeoutMinutes <= 0) { + return List.empty + } + + val now = currentTime() + val cutoff = new Timestamp(now.getTime - cleanupConfig.idleTimeoutMinutes * 60 * 1000) + val activeStatuses = Seq(Short.box(0), Short.box(1), Short.box(2)) + + withTransaction(context) { ctx => + val cuDao = new WorkflowComputingUnitDao(ctx.configuration()) + val userDao = new UserDao(ctx.configuration()) + ctx + .selectFrom(WORKFLOW_COMPUTING_UNIT) + .where( + WORKFLOW_COMPUTING_UNIT.TYPE + .eq(WorkflowComputingUnitTypeEnum.kubernetes) + .and(WORKFLOW_COMPUTING_UNIT.TERMINATE_TIME.isNull) + ) + .fetchInto(classOf[WorkflowComputingUnit]) + .asScala + .flatMap { unit => + val cuid = unit.getCuid + val hasActiveExecution = ctx.fetchExists( + ctx + .selectOne() + .from(WORKFLOW_EXECUTIONS) + .where( + WORKFLOW_EXECUTIONS.CUID + .eq(cuid) + .and(WORKFLOW_EXECUTIONS.STATUS.in(activeStatuses: _*)) + ) + ) + val latestUpdateTime = ctx + .select(max(WORKFLOW_EXECUTIONS.LAST_UPDATE_TIME)) + .from(WORKFLOW_EXECUTIONS) + .where(WORKFLOW_EXECUTIONS.CUID.eq(cuid)) + .fetchOne(0, classOf[Timestamp]) + val latestStartTime = ctx + .select(max(WORKFLOW_EXECUTIONS.STARTING_TIME)) + .from(WORKFLOW_EXECUTIONS) + .where(WORKFLOW_EXECUTIONS.CUID.eq(cuid)) + .fetchOne(0, classOf[Timestamp]) + val lastExecutionTime = lastComputingUnitActivityTime( + unit, + Option(latestUpdateTime), + Option(latestStartTime) + ) + + if (shouldTerminateIdleComputingUnit(hasActiveExecution, lastExecutionTime, cutoff)) { + if (podOperations.podExists(cuid)) { + podOperations.deletePod(cuid) + } + unit.setTerminateTime(now) + unit.setTerminationReason(WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED) + cuDao.update(unit) + val owner = Option(userDao.fetchOneByUid(unit.getUid)) + Some( + new TerminatedComputingUnitInfo( + cuid = unit.getCuid, + name = unit.getName, + uid = unit.getUid, + username = owner.flatMap(u => Option(u.getName).filter(_.nonEmpty)), + reason = WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED + ) + ) + } else { + None + } + } + .toList + } + } + private def icebergEnvironmentVariables: Map[String, Any] = { val base = Map[String, Any]( EnvironmentalVariable.ENV_ICEBERG_CATALOG_TYPE -> StorageConfig.icebergCatalogType @@ -131,6 +266,14 @@ object ComputingUnitManagingResource { .get ) + final class TerminatedComputingUnitInfo( + val cuid: Integer, + val name: String, + val uid: Integer, + val username: Option[String], + val reason: WorkflowComputingUnitTerminationReasonEnum + ) + case class WorkflowComputingUnitCreationParams( name: String, unitType: String, @@ -177,6 +320,7 @@ object ComputingUnitManagingResource { @Produces(Array(MediaType.APPLICATION_JSON)) @Path("/computing-unit") class ComputingUnitManagingResource { + private val logger = LoggerFactory.getLogger(classOf[ComputingUnitManagingService]) private def getComputingUnitByCuid(ctx: DSLContext, cuid: Int): WorkflowComputingUnit = { val wcDao = new WorkflowComputingUnitDao(ctx.configuration()) @@ -467,6 +611,12 @@ class ComputingUnitManagingResource { @Path("") def listComputingUnits( @Auth user: SessionUser + ): List[DashboardWorkflowComputingUnit] = + listComputingUnits(user, DefaultKubernetesPodOperations) + + private[resource] def listComputingUnits( + user: SessionUser, + podOperations: KubernetesPodOperations ): List[DashboardWorkflowComputingUnit] = { withTransaction(context) { ctx => val computingUnitDao = new WorkflowComputingUnitDao(ctx.configuration()) @@ -621,8 +771,13 @@ class ComputingUnitManagingResource { KubernetesClient.deletePod(cuid) } + val terminationReason = WorkflowComputingUnitTerminationReasonEnum.USER_REQUESTED unit.setTerminateTime(new Timestamp(System.currentTimeMillis())) + unit.setTerminationReason(terminationReason) cuDao.update(unit) + logger.info( + s"Terminated 1 Kubernetes computing unit(s): cuid=${unit.getCuid}, name=${unit.getName}, uid=${unit.getUid}, username=${user.getName}, reason=${terminationReason.getLiteral}" + ) } Response.ok().build() } diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala index 509603c2434..f2e6781cc84 100644 --- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala +++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala @@ -18,7 +18,10 @@ package org.apache.texera.service.util -import org.apache.texera.dao.jooq.generated.enums.WorkflowComputingUnitTypeEnum +import org.apache.texera.dao.jooq.generated.enums.{ + WorkflowComputingUnitTerminationReasonEnum, + WorkflowComputingUnitTypeEnum +} import org.apache.texera.dao.jooq.generated.tables.daos.{UserDao, WorkflowComputingUnitDao} import org.apache.texera.dao.jooq.generated.tables.pojos.WorkflowComputingUnit import org.apache.texera.service.resource.ComputingUnitManagingResource.{ @@ -200,7 +203,10 @@ object ComputingUnitHelpers { val vanished = partitioned._2 if (vanished.nonEmpty) { val now = new Timestamp(System.currentTimeMillis()) - vanished.foreach(_.setTerminateTime(now)) + vanished.foreach { unit => + unit.setTerminateTime(now) + unit.setTerminationReason(WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED) + } dao.update(vanished.asJava) } partitioned._1 diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/ComputingUnitIdleCleanupSchedulerSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/ComputingUnitIdleCleanupSchedulerSpec.scala new file mode 100644 index 00000000000..1138022e901 --- /dev/null +++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/ComputingUnitIdleCleanupSchedulerSpec.scala @@ -0,0 +1,237 @@ +/* + * 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.texera.service + +import io.dropwizard.core.setup.Environment +import io.dropwizard.lifecycle.setup.{LifecycleEnvironment, ScheduledExecutorServiceBuilder} +import org.apache.texera.dao.jooq.generated.enums.WorkflowComputingUnitTerminationReasonEnum +import org.apache.texera.service.resource.ComputingUnitManagingResource.TerminatedComputingUnitInfo +import org.mockito.Mockito.{mock, verify, when} +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +import java.util.concurrent.{ScheduledExecutorService, TimeUnit} + +class ComputingUnitIdleCleanupSchedulerSpec extends AnyFlatSpec with Matchers { + + "ComputingUnitManagingService.registerIdleComputingUnitCleanup" should "wire the Dropwizard scheduled executor" in { + val environment = mock(classOf[Environment]) + val lifecycle = mock(classOf[LifecycleEnvironment]) + val builder = mock(classOf[ScheduledExecutorServiceBuilder]) + val executor = mock(classOf[ScheduledExecutorService]) + val command = new Runnable { + override def run(): Unit = () + } + + when(environment.lifecycle()).thenReturn(lifecycle) + when(lifecycle.scheduledExecutorService("idle-computing-unit-terminator")).thenReturn(builder) + when(builder.threads(1)).thenReturn(builder) + when(builder.build()).thenReturn(executor) + + val service = new ComputingUnitManagingService { + override private[service] def registerIdleComputingUnitCleanup( + scheduleWithFixedDelay: (Runnable, Long, Long, TimeUnit) => Unit + ): Unit = + scheduleWithFixedDelay(command, 7, 11, TimeUnit.SECONDS) + } + + service.registerIdleComputingUnitCleanup(environment) + + verify(executor).scheduleWithFixedDelay(command, 7, 11, TimeUnit.SECONDS) + } + + it should "schedule cleanup only when Kubernetes idle cleanup is enabled" in { + var scheduled = false + + ComputingUnitManagingService.registerIdleComputingUnitCleanup( + (_, _, _, _) => scheduled = true, + kubernetesComputingUnitEnabled = false, + idleTimeoutMinutes = 60, + idleCheckIntervalMinutes = 15, + terminateIdleComputingUnits = () => List.empty, + logTerminatedUnits = _ => (), + logCleanupFailure = _ => () + ) + scheduled shouldBe false + + ComputingUnitManagingService.registerIdleComputingUnitCleanup( + (_, _, _, _) => scheduled = true, + kubernetesComputingUnitEnabled = true, + idleTimeoutMinutes = 60, + idleCheckIntervalMinutes = 15, + terminateIdleComputingUnits = () => List.empty, + logTerminatedUnits = _ => (), + logCleanupFailure = _ => () + ) + scheduled shouldBe true + } + + "ComputingUnitManagingService.scheduleIdleComputingUnitCleanup" should "schedule cleanup with the configured fixed delay" in { + var scheduledCommand: Runnable = null + var scheduledInitialDelay: Long = -1 + var scheduledDelay: Long = -1 + var scheduledTimeUnit: TimeUnit = null + var cleanupInvocations = 0 + var infoMessages = List.empty[String] + var failures = List.empty[Throwable] + + ComputingUnitManagingService.scheduleIdleComputingUnitCleanup( + (command, initialDelay, delay, unit) => { + scheduledCommand = command + scheduledInitialDelay = initialDelay + scheduledDelay = delay + scheduledTimeUnit = unit + }, + idleCheckIntervalMinutes = 15, + terminateIdleComputingUnits = () => { + cleanupInvocations += 1 + List.empty + }, + logTerminatedUnits = message => infoMessages = infoMessages :+ message, + logCleanupFailure = throwable => failures = failures :+ throwable + ) + + scheduledCommand should not be null + scheduledInitialDelay shouldBe 15 + scheduledDelay shouldBe 15 + scheduledTimeUnit shouldBe TimeUnit.MINUTES + + scheduledCommand.run() + + cleanupInvocations shouldBe 1 + infoMessages shouldBe empty + failures shouldBe empty + } + + it should "run the scheduled cleanup command and log terminated units" in { + var scheduledCommand: Runnable = null + var cleanupInvocations = 0 + var infoMessages = List.empty[String] + var failures = List.empty[Throwable] + + ComputingUnitManagingService.scheduleIdleComputingUnitCleanup( + (command, _, _, _) => scheduledCommand = command, + idleCheckIntervalMinutes = 5, + terminateIdleComputingUnits = () => { + cleanupInvocations += 1 + List( + new TerminatedComputingUnitInfo( + cuid = 3, + name = "scheduled-idle", + uid = 30, + username = Some("carol"), + reason = WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED + ) + ) + }, + logTerminatedUnits = message => infoMessages = infoMessages :+ message, + logCleanupFailure = throwable => failures = failures :+ throwable + ) + + scheduledCommand.run() + + cleanupInvocations shouldBe 1 + infoMessages shouldBe List( + "Terminated 1 idle Kubernetes computing unit(s): " + + "cuid=3, name=scheduled-idle, uid=30, username=carol, reason=GARBAGE_COLLECTED" + ) + failures shouldBe empty + } + + "ComputingUnitManagingService.runIdleComputingUnitCleanup" should "not log when no idle computing units are terminated" in { + var infoMessages = List.empty[String] + var failures = List.empty[Throwable] + + ComputingUnitManagingService.runIdleComputingUnitCleanup( + () => List.empty, + message => infoMessages = infoMessages :+ message, + throwable => failures = failures :+ throwable + ) + + infoMessages shouldBe empty + failures shouldBe empty + } + + it should "log terminated idle computing unit details" in { + var infoMessages = List.empty[String] + var failures = List.empty[Throwable] + + ComputingUnitManagingService.runIdleComputingUnitCleanup( + () => + List( + new TerminatedComputingUnitInfo( + cuid = 1, + name = "idle-a", + uid = 10, + username = Some("alice"), + reason = WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED + ), + new TerminatedComputingUnitInfo( + cuid = 2, + name = "idle-b", + uid = 20, + username = None, + reason = WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED + ) + ), + message => infoMessages = infoMessages :+ message, + throwable => failures = failures :+ throwable + ) + + infoMessages shouldBe List( + "Terminated 2 idle Kubernetes computing unit(s): " + + "cuid=1, name=idle-a, uid=10, username=alice, reason=GARBAGE_COLLECTED; " + + "cuid=2, name=idle-b, uid=20, username=unknown, reason=GARBAGE_COLLECTED" + ) + failures shouldBe empty + } + + it should "log cleanup failures without throwing" in { + val failure = new RuntimeException("cleanup failed") + var infoMessages = List.empty[String] + var failures = List.empty[Throwable] + + noException shouldBe thrownBy { + ComputingUnitManagingService.runIdleComputingUnitCleanup( + () => throw failure, + message => infoMessages = infoMessages :+ message, + throwable => failures = failures :+ throwable + ) + } + + infoMessages shouldBe empty + failures shouldBe List(failure) + } + + "ComputingUnitManagingService.shouldScheduleIdleComputingUnitCleanup" should "enable scheduling only when Kubernetes cleanup is enabled with a positive timeout" in { + ComputingUnitManagingService.shouldScheduleIdleComputingUnitCleanup( + kubernetesComputingUnitEnabled = true, + idleTimeoutMinutes = 1 + ) shouldBe true + ComputingUnitManagingService.shouldScheduleIdleComputingUnitCleanup( + kubernetesComputingUnitEnabled = false, + idleTimeoutMinutes = 1 + ) shouldBe false + ComputingUnitManagingService.shouldScheduleIdleComputingUnitCleanup( + kubernetesComputingUnitEnabled = true, + idleTimeoutMinutes = 0 + ) shouldBe false + } +} diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitIdleCleanupSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitIdleCleanupSpec.scala new file mode 100644 index 00000000000..ecb6baae71e --- /dev/null +++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitIdleCleanupSpec.scala @@ -0,0 +1,425 @@ +/* + * 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.texera.service.resource + +import org.apache.texera.auth.SessionUser +import org.apache.texera.dao.MockTexeraDB +import org.apache.texera.dao.jooq.generated.Tables.{ + USER => USER_TABLE, + WORKFLOW, + WORKFLOW_COMPUTING_UNIT, + WORKFLOW_EXECUTIONS, + WORKFLOW_VERSION +} +import org.apache.texera.dao.jooq.generated.enums.{ + WorkflowComputingUnitTerminationReasonEnum, + WorkflowComputingUnitTypeEnum +} +import org.apache.texera.dao.jooq.generated.tables.daos.{ + UserDao, + WorkflowComputingUnitDao, + WorkflowDao, + WorkflowExecutionsDao, + WorkflowVersionDao +} +import org.apache.texera.dao.jooq.generated.tables.pojos.{ + User, + Workflow, + WorkflowComputingUnit, + WorkflowExecutions, + WorkflowVersion +} +import org.apache.texera.service.resource.ComputingUnitManagingResource.{ + IdleComputingUnitCleanupConfig, + KubernetesPodOperations, + TerminatedComputingUnitInfo +} +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers +import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach} + +import java.sql.Timestamp +import java.util.UUID +import java.util.concurrent.TimeUnit + +class ComputingUnitIdleCleanupSpec + extends AnyFlatSpec + with Matchers + with BeforeAndAfterAll + with BeforeAndAfterEach + with MockTexeraDB { + + private val testUserId = 810000 + scala.util.Random.nextInt(10000) + private val testWorkflowId = 820000 + scala.util.Random.nextInt(10000) + private val now = new Timestamp(TimeUnit.DAYS.toMillis(20)) + private val cleanupConfig = + new IdleComputingUnitCleanupConfig(enabled = true, idleTimeoutMinutes = 60) + + private var userDao: UserDao = _ + private var workflowDao: WorkflowDao = _ + private var workflowVersionDao: WorkflowVersionDao = _ + private var workflowComputingUnitDao: WorkflowComputingUnitDao = _ + private var workflowExecutionsDao: WorkflowExecutionsDao = _ + private var testVersion: WorkflowVersion = _ + + override protected def beforeAll(): Unit = + initializeDBAndReplaceDSLContext() + + override protected def beforeEach(): Unit = { + userDao = new UserDao(getDSLContext.configuration()) + workflowDao = new WorkflowDao(getDSLContext.configuration()) + workflowVersionDao = new WorkflowVersionDao(getDSLContext.configuration()) + workflowComputingUnitDao = new WorkflowComputingUnitDao(getDSLContext.configuration()) + workflowExecutionsDao = new WorkflowExecutionsDao(getDSLContext.configuration()) + + cleanupTestData() + + val user = new User + user.setUid(testUserId) + user.setName("idle-cu-owner") + user.setEmail(s"idle-cu-${UUID.randomUUID()}@example.com") + userDao.insert(user) + + val workflow = new Workflow + workflow.setWid(testWorkflowId) + workflow.setName("idle-cu-workflow") + workflow.setContent("{}") + workflow.setCreationTime(new Timestamp(now.getTime - TimeUnit.DAYS.toMillis(2))) + workflow.setLastModifiedTime(new Timestamp(now.getTime - TimeUnit.DAYS.toMillis(2))) + workflowDao.insert(workflow) + + testVersion = new WorkflowVersion + testVersion.setWid(testWorkflowId) + testVersion.setContent("{}") + testVersion.setCreationTime(new Timestamp(now.getTime - TimeUnit.DAYS.toMillis(2))) + workflowVersionDao.insert(testVersion) + } + + override protected def afterEach(): Unit = + cleanupTestData() + + override protected def afterAll(): Unit = + shutdownDB() + + private def cleanupTestData(): Unit = { + getDSLContext + .deleteFrom(WORKFLOW_EXECUTIONS) + .where(WORKFLOW_EXECUTIONS.UID.eq(testUserId)) + .execute() + getDSLContext + .deleteFrom(WORKFLOW_COMPUTING_UNIT) + .where(WORKFLOW_COMPUTING_UNIT.UID.eq(testUserId)) + .execute() + getDSLContext + .deleteFrom(WORKFLOW_VERSION) + .where(WORKFLOW_VERSION.WID.eq(testWorkflowId)) + .execute() + getDSLContext.deleteFrom(WORKFLOW).where(WORKFLOW.WID.eq(testWorkflowId)).execute() + getDSLContext.deleteFrom(USER_TABLE).where(USER_TABLE.UID.eq(testUserId)).execute() + } + + private def timestampMinutesBefore(minutes: Long): Timestamp = + new Timestamp(now.getTime - TimeUnit.MINUTES.toMillis(minutes)) + + private def insertComputingUnit( + name: String, + unitType: WorkflowComputingUnitTypeEnum = WorkflowComputingUnitTypeEnum.kubernetes, + creationMinutesBefore: Long = 120, + terminated: Boolean = false + ): WorkflowComputingUnit = { + val unit = new WorkflowComputingUnit + unit.setUid(testUserId) + unit.setName(name) + unit.setCreationTime(timestampMinutesBefore(creationMinutesBefore)) + unit.setType(unitType) + unit.setUri("kubernetes://test") + unit.setResource("{}") + if (terminated) { + unit.setTerminateTime(timestampMinutesBefore(10)) + unit.setTerminationReason(WorkflowComputingUnitTerminationReasonEnum.USER_REQUESTED) + } + workflowComputingUnitDao.insert(unit) + unit + } + + private def insertExecution( + unit: WorkflowComputingUnit, + status: Short, + startingMinutesBefore: Long, + lastUpdateMinutesBefore: Option[Long] = None + ): Unit = { + val execution = new WorkflowExecutions + execution.setVid(testVersion.getVid) + execution.setUid(testUserId) + execution.setCuid(unit.getCuid) + execution.setStatus(status) + execution.setStartingTime(timestampMinutesBefore(startingMinutesBefore)) + lastUpdateMinutesBefore.foreach(minutes => + execution.setLastUpdateTime(timestampMinutesBefore(minutes)) + ) + execution.setBookmarked(false) + execution.setName("execution-" + UUID.randomUUID().toString.substring(0, 8)) + execution.setEnvironmentVersion("test-env") + workflowExecutionsDao.insert(execution) + } + + private def sessionUser( + uid: Integer = testUserId, + name: String = "idle-cu-owner" + ): SessionUser = { + val user = new User + user.setUid(uid) + user.setName(name) + new SessionUser(user) + } + + private class RecordingPodOperations(existingPods: Set[Int]) extends KubernetesPodOperations { + var deletedPods: List[Int] = List.empty + + override val podExists: Int => Boolean = cuid => existingPods.contains(cuid) + override val deletePod: Int => Unit = cuid => deletedPods = deletedPods :+ cuid + } + + "cleanup value objects" should "expose cleanup configuration and termination info fields" in { + val config = new IdleComputingUnitCleanupConfig(enabled = true, idleTimeoutMinutes = 60) + config.copy(enabled = false).enabled shouldBe false + config.copy(idleTimeoutMinutes = 30).idleTimeoutMinutes shouldBe 30 + + val terminated = new TerminatedComputingUnitInfo( + cuid = 1, + name = "plain-unit-info", + uid = testUserId, + username = Some("idle-cu-owner"), + reason = WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED + ) + terminated.cuid shouldBe 1 + terminated.username shouldBe Some("idle-cu-owner") + terminated.reason shouldBe WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED + } + + "terminateIdleKubernetesComputingUnits" should "return empty without scanning when cleanup is disabled" in { + val podOperations = new RecordingPodOperations(Set.empty) + + ComputingUnitManagingResource.terminateIdleKubernetesComputingUnits( + cleanupConfig.copy(enabled = false), + () => now, + podOperations + ) shouldBe empty + + podOperations.deletedPods shouldBe empty + } + + it should "return empty when the idle timeout is disabled" in { + val podOperations = new RecordingPodOperations(Set.empty) + + ComputingUnitManagingResource.terminateIdleKubernetesComputingUnits( + cleanupConfig.copy(idleTimeoutMinutes = 0), + () => now, + podOperations + ) shouldBe empty + + podOperations.deletedPods shouldBe empty + } + + it should "garbage collect only inactive Kubernetes computing units past the timeout" in { + val stale = insertComputingUnit("stale") + val active = insertComputingUnit("active") + val recent = insertComputingUnit("recent") + val local = insertComputingUnit("local", WorkflowComputingUnitTypeEnum.local) + val alreadyTerminated = insertComputingUnit("already-terminated", terminated = true) + + insertExecution(active, status = 1, startingMinutesBefore = 180) + insertExecution( + recent, + status = 3, + startingMinutesBefore = 180, + lastUpdateMinutesBefore = Some(5) + ) + insertExecution(local, status = 3, startingMinutesBefore = 180) + + val podOperations = new RecordingPodOperations(Set(stale.getCuid, active.getCuid)) + val terminated = ComputingUnitManagingResource.terminateIdleKubernetesComputingUnits( + cleanupConfig, + () => now, + podOperations + ) + + terminated.map(_.cuid) shouldBe List(stale.getCuid) + terminated.head.username shouldBe Some("idle-cu-owner") + terminated.head.reason shouldBe WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED + podOperations.deletedPods shouldBe List(stale.getCuid) + + val staleAfterCleanup = workflowComputingUnitDao.fetchOneByCuid(stale.getCuid) + staleAfterCleanup.getTerminateTime shouldBe now + staleAfterCleanup.getTerminationReason shouldBe WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED + workflowComputingUnitDao.fetchOneByCuid(active.getCuid).getTerminateTime shouldBe null + workflowComputingUnitDao.fetchOneByCuid(recent.getCuid).getTerminateTime shouldBe null + workflowComputingUnitDao.fetchOneByCuid(local.getCuid).getTerminateTime shouldBe null + workflowComputingUnitDao.fetchOneByCuid(alreadyTerminated.getCuid).getTerminationReason shouldBe + WorkflowComputingUnitTerminationReasonEnum.USER_REQUESTED + } + + it should "mark an idle unit terminated even when the pod is already absent" in { + val stale = insertComputingUnit("stale-missing-pod") + val podOperations = new RecordingPodOperations(Set.empty) + + val terminated = ComputingUnitManagingResource.terminateIdleKubernetesComputingUnits( + cleanupConfig, + () => now, + podOperations + ) + + terminated.map(_.cuid) shouldBe List(stale.getCuid) + podOperations.deletedPods shouldBe empty + workflowComputingUnitDao.fetchOneByCuid(stale.getCuid).getTerminationReason shouldBe + WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED + } + + it should "keep units with active status 0 or 2 running" in { + val activeQueued = insertComputingUnit("active-status-0") + val activeRunning = insertComputingUnit("active-status-2") + insertExecution(activeQueued, status = 0, startingMinutesBefore = 180) + insertExecution(activeRunning, status = 2, startingMinutesBefore = 180) + val podOperations = new RecordingPodOperations(Set(activeQueued.getCuid, activeRunning.getCuid)) + + ComputingUnitManagingResource.terminateIdleKubernetesComputingUnits( + cleanupConfig, + () => now, + podOperations + ) shouldBe empty + + podOperations.deletedPods shouldBe empty + workflowComputingUnitDao.fetchOneByCuid(activeQueued.getCuid).getTerminateTime shouldBe null + workflowComputingUnitDao.fetchOneByCuid(activeRunning.getCuid).getTerminateTime shouldBe null + } + + it should "terminate a unit whose latest completed execution activity is past the timeout" in { + val staleWithExecution = insertComputingUnit("stale-completed-execution") + insertExecution( + staleWithExecution, + status = 3, + startingMinutesBefore = 180, + lastUpdateMinutesBefore = Some(90) + ) + val podOperations = new RecordingPodOperations(Set(staleWithExecution.getCuid)) + + val terminated = ComputingUnitManagingResource.terminateIdleKubernetesComputingUnits( + cleanupConfig, + () => now, + podOperations + ) + + terminated.map(_.cuid) shouldBe List(staleWithExecution.getCuid) + podOperations.deletedPods shouldBe List(staleWithExecution.getCuid) + workflowComputingUnitDao + .fetchOneByCuid(staleWithExecution.getCuid) + .getTerminationReason shouldBe + WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED + } + + it should "omit an empty owner name from terminated unit info" in { + val user = userDao.fetchOneByUid(testUserId) + user.setName("") + userDao.update(user) + insertComputingUnit("stale-empty-owner") + val podOperations = new RecordingPodOperations(Set.empty) + + val terminated = ComputingUnitManagingResource.terminateIdleKubernetesComputingUnits( + cleanupConfig, + () => now, + podOperations + ) + + terminated should have size 1 + terminated.head.username shouldBe None + } + + "terminateComputingUnit" should "mark manual termination as user requested" in { + val local = insertComputingUnit("manual-local", WorkflowComputingUnitTypeEnum.local) + + val response = + new ComputingUnitManagingResource().terminateComputingUnit(local.getCuid, sessionUser()) + + response.getStatus shouldBe 200 + val terminated = workflowComputingUnitDao.fetchOneByCuid(local.getCuid) + terminated.getTerminateTime should not be null + terminated.getTerminationReason shouldBe WorkflowComputingUnitTerminationReasonEnum.USER_REQUESTED + } + + it should "reject manual termination from a non-owner" in { + val local = insertComputingUnit("manual-local-non-owner", WorkflowComputingUnitTypeEnum.local) + + val response = new ComputingUnitManagingResource().terminateComputingUnit( + local.getCuid, + sessionUser(uid = testUserId + 1) + ) + + response.getStatus shouldBe 400 + workflowComputingUnitDao.fetchOneByCuid(local.getCuid).getTerminateTime shouldBe null + } + + "lastComputingUnitActivityTime" should "prefer the latest execution timestamp over creation time" in { + val unit = new WorkflowComputingUnit + unit.setCreationTime(timestampMinutesBefore(120)) + + ComputingUnitManagingResource.lastComputingUnitActivityTime( + unit, + latestUpdateTime = Some(timestampMinutesBefore(10)), + latestStartTime = Some(timestampMinutesBefore(30)) + ) shouldBe timestampMinutesBefore(10) + } + + it should "fall back to start time and then creation time" in { + val unit = new WorkflowComputingUnit + unit.setCreationTime(timestampMinutesBefore(120)) + + ComputingUnitManagingResource.lastComputingUnitActivityTime( + unit, + latestUpdateTime = None, + latestStartTime = Some(timestampMinutesBefore(30)) + ) shouldBe timestampMinutesBefore(30) + + ComputingUnitManagingResource.lastComputingUnitActivityTime( + unit, + latestUpdateTime = None, + latestStartTime = None + ) shouldBe timestampMinutesBefore(120) + } + + "shouldTerminateIdleComputingUnit" should "require both no active execution and activity before cutoff" in { + val cutoff = timestampMinutesBefore(60) + + ComputingUnitManagingResource.shouldTerminateIdleComputingUnit( + hasActiveExecution = false, + lastExecutionTime = timestampMinutesBefore(61), + cutoff = cutoff + ) shouldBe true + ComputingUnitManagingResource.shouldTerminateIdleComputingUnit( + hasActiveExecution = true, + lastExecutionTime = timestampMinutesBefore(61), + cutoff = cutoff + ) shouldBe false + ComputingUnitManagingResource.shouldTerminateIdleComputingUnit( + hasActiveExecution = false, + lastExecutionTime = timestampMinutesBefore(60), + cutoff = cutoff + ) shouldBe false + } +} diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala index 5cf6fae2146..187672b969e 100644 --- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala +++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala @@ -317,6 +317,8 @@ class ComputingUnitHelpersSpec live.map(_.getCuid) should contain theSameElementsAs Seq(600, 602) computingUnitDao.fetchOneByCuid(601).getTerminateTime should not be null + computingUnitDao.fetchOneByCuid(601).getTerminationReason shouldBe + WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED computingUnitDao.fetchOneByCuid(600).getTerminateTime shouldBe null computingUnitDao.fetchOneByCuid(602).getTerminateTime shouldBe null } diff --git a/sql/texera_ddl.sql b/sql/texera_ddl.sql index b8ebe3caf6d..e21af07c2f5 100644 --- a/sql/texera_ddl.sql +++ b/sql/texera_ddl.sql @@ -95,6 +95,7 @@ CREATE TYPE user_role_enum AS ENUM ('INACTIVE', 'RESTRICTED', 'REGULAR', 'ADMIN' CREATE TYPE action_enum AS ENUM ('like', 'unlike', 'view', 'clone'); CREATE TYPE privilege_enum AS ENUM ('NONE', 'READ', 'WRITE'); CREATE TYPE workflow_computing_unit_type_enum AS ENUM ('local', 'kubernetes'); +CREATE TYPE workflow_computing_unit_termination_reason_enum AS ENUM ('USER_REQUESTED', 'GARBAGE_COLLECTED'); CREATE TYPE provider_type_enum AS ENUM ('LOCAL', 'GOOGLE'); CREATE TYPE user_warehouse_flavor_enum AS ENUM ('local', 'aws'); @@ -247,6 +248,7 @@ CREATE TABLE IF NOT EXISTS workflow_computing_unit cuid SERIAL PRIMARY KEY, creation_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, terminate_time TIMESTAMP DEFAULT NULL, + termination_reason workflow_computing_unit_termination_reason_enum DEFAULT NULL, type workflow_computing_unit_type_enum, uri TEXT NOT NULL DEFAULT '', resource TEXT DEFAULT '', diff --git a/sql/updates/38.sql b/sql/updates/38.sql new file mode 100644 index 00000000000..7d95046b945 --- /dev/null +++ b/sql/updates/38.sql @@ -0,0 +1,41 @@ +/* + * 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. + */ + +\c texera_db + +SET search_path TO texera_db; + +BEGIN; + +DO $$ +BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_type WHERE typname = 'workflow_computing_unit_termination_reason_enum' + ) THEN + CREATE TYPE workflow_computing_unit_termination_reason_enum AS ENUM ( + 'USER_REQUESTED', + 'GARBAGE_COLLECTED' + ); + END IF; +END $$; + +ALTER TABLE workflow_computing_unit + ADD COLUMN IF NOT EXISTS termination_reason workflow_computing_unit_termination_reason_enum DEFAULT NULL; + +COMMIT;