@ -4,16 +4,22 @@ import kr.re.etri.autoflow.payload.request.CreateRunRequest;
import kr.re.etri.autoflow.payload.request.RunCreatedEvent ;
import lombok.RequiredArgsConstructor ;
import lombok.extern.slf4j.Slf4j ;
import org.springframework.batch.core.Job ;
import org.springframework.batch.core.JobParametersBuilder ;
import org.springframework.batch.core.launch.JobLauncher ;
import org.springframework.beans.factory.annotation.Autowired ;
import org.springframework.beans.factory.annotation.Value ;
import org.springframework.context.ApplicationEventPublisher ;
import org.springframework.core.io.InputStreamResource ;
import org.springframework.http.* ;
import org.springframework.http.HttpEntity ;
import org.springframework.http.HttpHeaders ;
import org.springframework.http.MediaType ;
import org.springframework.http.ResponseEntity ;
import org.springframework.stereotype.Service ;
import org.springframework.util.LinkedMultiValueMap ;
import org.springframework.util.MultiValueMap ;
import org.springframework.web.client.RestTemplate ;
import org.springframework.web.multipart.MultipartFile ;
import org.springframework.web.reactive.function.BodyInserters ;
import org.springframework.web.reactive.function.client.WebClient ;
import org.springframework.web.reactive.function.client.WebClientResponseException ;
import org.springframework.web.util.UriComponentsBuilder ;
@ -21,10 +27,13 @@ import org.springframework.web.util.UriUtils;
import reactor.core.publisher.Mono ;
import java.io.IOException ;
import java.net.URI ;
import java.nio.charset.StandardCharsets ;
import java.time.Duration ;
import java.util.* ;
import java.util.Collections ;
import java.util.HashMap ;
import java.util.HashSet ;
import java.util.Map ;
import java.util.Set ;
import java.util.concurrent.CompletableFuture ;
@Service
@ -34,629 +43,323 @@ public class PipelineUploadService {
private final RestTemplate restTemplate ;
private final WebClient webClient ;
@Value ( "${kubeflow.url}" )
private String kubeflowBaseUrl ;
private String kubeflowBaseUrl ; // 예: http://192.168.10.135:32473/
private final WebClient webClient ;
@Autowired
private ApplicationEventPublisher eventPublisher ;
/ * *
* Pipeline 업 로 드
* /
public Map uploadPipeline (
MultipartFile file ,
String name ,
String displayName ,
String description ,
String namespace
) {
public Map uploadPipeline ( MultipartFile file ,
String name ,
String displayName ,
String description ,
String namespace ) {
try {
log . info ( "" "
= = = = = Pipeline Upload Start = = = = =
filename = { }
name = { }
displayName = { }
description = { }
namespace = { }
"" " ,
file . getOriginalFilename ( ) ,
name ,
displayName ,
description ,
namespace
) ;
MultiValueMap < String , Object > body =
new LinkedMultiValueMap < > ( ) ;
body . add (
"uploadfile" ,
new MultipartInputStreamFileResource (
file . getInputStream ( ) ,
file . getOriginalFilename ( )
)
) ;
MultiValueMap < String , Object > body = new LinkedMultiValueMap < > ( ) ;
body . add ( "uploadfile" , new MultipartInputStreamFileResource ( file . getInputStream ( ) , file . getOriginalFilename ( ) ) ) ;
HttpHeaders headers = new HttpHeaders ( ) ;
headers . setContentType (
MediaType . MULTIPART_FORM_DATA
) ;
headers . setContentType ( MediaType . MULTIPART_FORM_DATA ) ;
HttpEntity < MultiValueMap < String , Object > >
requestEntity =
new HttpEntity < > ( body , headers ) ;
HttpEntity < MultiValueMap < String , Object > > requestEntity = new HttpEntity < > ( body , headers ) ;
URI uri = UriComponentsBuilder
. fromHttpUrl (
normalizeBaseUrl (
kubeflowBaseUrl
)
)
. path ( "/apis/v2beta1/pipelines/upload" )
. queryParamIfPresent (
"name" ,
optional ( name )
)
. queryParamIfPresent (
"display_name" ,
optional ( displayName )
)
. queryParamIfPresent (
"description" ,
optional ( description )
)
. queryParamIfPresent (
"namespace" ,
optional ( namespace )
)
. build ( true )
. toUri ( ) ;
UriComponentsBuilder builder = UriComponentsBuilder . fromUriString ( kubeflowBaseUrl . replaceAll ( "/+$" , "" ) + "/apis/v2beta1/pipelines/upload" ) ;
log . info ( "Pipeline Upload URI={}" , uri ) ;
ResponseEntity < Map > response =
restTemplate . postForEntity (
uri ,
requestEntity ,
Map . class
) ;
if ( name ! = null & & ! name . isBlank ( ) ) builder . queryParam ( "name" , name ) ;
if ( displayName ! = null & & ! displayName . isBlank ( ) ) builder . queryParam ( "display_name" , displayName ) ;
if ( description ! = null & & ! description . isBlank ( ) ) builder . queryParam ( "description" , description ) ;
if ( namespace ! = null & & ! namespace . isBlank ( ) ) builder . queryParam ( "namespace" , namespace ) ;
ResponseEntity < Map > response = restTemplate . postForEntity ( builder . toUriString ( ) , requestEntity , Map . class ) ;
return response . getBody ( ) ;
} catch ( IOException e ) {
log . error (
"Pipeline upload failed" ,
e
) ;
throw new RuntimeException (
"Pipeline upload failed" ,
e
) ;
throw new RuntimeException ( "Pipeline upload failed" , e ) ;
}
}
/ * *
* Run 생 성
* runRequest 에 는 display_name , pipeline_version_reference , runtime_config 등 이 포 함 되 어 야 함 .
* KFP v2beta1 는 runtime_config 필 수 , display_name 필 수 .
* parameters 는 파 이 프 라 인 스 펙 에 정 의 된 입 력 만 전 달 하 며 , mlflow_experiment_name 은 파 이 프 라 인 에 정 의 돼 있 을 때 만 포 함 한 다 .
* /
public Map < String , Object > createRun (
CreateRunRequest runRequest
) {
Set < String > allowedParamNames =
getPipelineRootParameterNames (
runRequest
) ;
Map < String , Object > body =
buildKfpRunRequestBody (
runRequest ,
allowedParamNames
) ;
String uri =
normalizeBaseUrl ( kubeflowBaseUrl )
+ "/apis/v2beta1/runs" ;
log . info ( "Create Run URI={}" , uri ) ;
public Map < String , Object > createRun ( CreateRunRequest runRequest ) {
Set < String > allowedParamNames = getPipelineRootParameterNames ( runRequest ) ;
Map < String , Object > body = buildKfpRunRequestBody ( runRequest , allowedParamNames ) ;
log . debug ( "[KFP] CreateRun request body: {}" , body ) ;
Map result = webClient . post ( )
. uri ( uri )
. contentType (
MediaType . APPLICATION_JSON
)
. uri ( kubeflowBaseUrl . replaceAll ( "/+$" , "" ) + "/apis/v2beta1/runs" )
. contentType ( MediaType . APPLICATION_JSON )
. bodyValue ( body )
. retrieve ( )
. onStatus (
status - >
status . is4xxClientError ( )
| | status . is5xxServerError ( ) ,
response - >
response . bodyToMono ( String . class )
. map ( msg - >
new RuntimeException (
"KFP CreateRun failed: "
+ response . statusCode ( )
+ " "
+ msg
)
)
)
. onStatus ( status - > status . is4xxClientError ( ) | | status . is5xxServerError ( ) ,
resp - > resp . bodyToMono ( String . class )
. doOnNext ( msg - > log . warn ( "[KFP] CreateRun error {}: {}" , resp . statusCode ( ) , msg ) )
. map ( msg - > new RuntimeException ( "KFP CreateRun failed: " + resp . statusCode ( ) + " " + msg ) ) )
. bodyToMono ( Map . class )
. block ( ) ;
if ( result ! = null & &
result . get ( "run_id" ) ! = null ) {
String runId =
( String ) result . get ( "run_id" ) ;
CompletableFuture . runAsync ( ( ) - >
eventPublisher . publishEvent (
new RunCreatedEvent ( runId )
)
) ;
// 이벤트 발행만 비동기로 처리
if ( result ! = null & & result . get ( "run_id" ) ! = null ) {
String runId = ( String ) result . get ( "run_id" ) ;
CompletableFuture . runAsync ( ( ) - > eventPublisher . publishEvent ( new RunCreatedEvent ( runId ) ) ) ;
}
return result ;
}
/ * *
* Pipeline Version Spec 조 회
* KFP pipeline version 스 펙 을 조 회 하 여 root 입 력 파 라 미 터 이 름 목 록 을 반 환 한 다 .
* pipeline_version_reference 에 pipeline_id , pipeline_version_id 가 있 을 때 만 조 회 하 며 , 실 패 시 null 반 환 .
* /
@SuppressWarnings ( "unchecked" )
private Set < String > getPipelineRootParameterNames (
CreateRunRequest runRequest
) {
CreateRunRequest . PipelineVersionReference ref =
runRequest ! = null
? runRequest . getPipeline_version_reference ( )
: null ;
if ( ref = = null
| | ref . getPipeline_id ( ) = = null
| | ref . getPipeline_version_id ( ) = = null ) {
private Set < String > getPipelineRootParameterNames ( CreateRunRequest runRequest ) {
CreateRunRequest . PipelineVersionReference ref = runRequest ! = null ? runRequest . getPipeline_version_reference ( ) : null ;
if ( ref = = null | | ref . getPipeline_id ( ) = = null | | ref . getPipeline_id ( ) . isBlank ( )
| | ref . getPipeline_version_id ( ) = = null | | ref . getPipeline_version_id ( ) . isBlank ( ) ) {
return null ;
}
String url =
normalizeBaseUrl ( kubeflowBaseUrl )
+ "/apis/v2beta1/pipelines/"
+ ref . getPipeline_id ( )
+ "/versions/"
+ ref . getPipeline_version_id ( ) ;
String url = kubeflowBaseUrl . replaceAll ( "/+$" , "" ) + "/apis/v2beta1/pipelines/" + ref . getPipeline_id ( ) + "/versions/" + ref . getPipeline_version_id ( ) ;
try {
Map < String , Object > version =
webClient . get ( )
. uri ( url )
. retrieve ( )
. bodyToMono ( Map . class )
. block ( ) ;
if ( version = = null ) {
return null ;
}
Map < String , Object > spec =
( Map < String , Object > )
version . get ( "pipeline_spec" ) ;
if ( spec = = null ) {
return null ;
}
Map < String , Object > root =
( Map < String , Object > )
spec . get ( "root" ) ;
if ( root = = null ) {
root =
( Map < String , Object > )
spec . get ( "Root" ) ;
}
if ( root = = null ) {
return null ;
}
Map < String , Object > inputDefinitions =
( Map < String , Object > )
root . get ( "inputDefinitions" ) ;
if ( inputDefinitions = = null ) {
inputDefinitions =
( Map < String , Object > )
root . get ( "input_definitions" ) ;
}
if ( inputDefinitions = = null ) {
return null ;
}
Object parameters =
inputDefinitions . get ( "parameters" ) ;
if ( ! ( parameters instanceof Map < ? , ? > map ) ) {
return Collections . emptySet ( ) ;
}
Map < String , Object > version = webClient . get ( )
. uri ( url )
. accept ( MediaType . APPLICATION_JSON )
. retrieve ( )
. bodyToMono ( Map . class )
. block ( ) ;
if ( version = = null ) return null ;
Object spec = version . get ( "pipeline_spec" ) ;
if ( ! ( spec instanceof Map ) ) return null ;
Map < String , Object > specMap = ( Map < String , Object > ) spec ;
Object root = specMap . get ( "root" ) ;
if ( root = = null ) root = specMap . get ( "Root" ) ;
if ( ! ( root instanceof Map ) ) return null ;
Map < String , Object > rootMap = ( Map < String , Object > ) root ;
Object inputDef = rootMap . get ( "inputDefinitions" ) ;
if ( inputDef = = null ) inputDef = rootMap . get ( "input_definitions" ) ;
if ( ! ( inputDef instanceof Map ) ) return null ;
Map < String , Object > inputDefMap = ( Map < String , Object > ) inputDef ;
Object params = inputDefMap . get ( "parameters" ) ;
if ( params = = null ) return Collections . emptySet ( ) ;
Set < String > names = new HashSet < > ( ) ;
for ( Object key : map . keySet ( ) ) {
names . add ( key . toString ( ) ) ;
if ( params instanceof Map ) {
for ( Object key : ( ( Map < ? , ? > ) params ) . keySet ( ) ) {
if ( key ! = null ) names . add ( key . toString ( ) ) ;
}
return names ;
}
return names ;
if ( params instanceof Iterable ) {
for ( Object item : ( Iterable < ? > ) params ) {
if ( item instanceof Map ) {
Object name = ( ( Map < ? , ? > ) item ) . get ( "name" ) ;
if ( name ! = null ) names . add ( name . toString ( ) ) ;
}
}
return names ;
}
return Collections . emptySet ( ) ;
} catch ( Exception e ) {
log . warn (
"Pipeline version spec 조회 실패" ,
e
) ;
log . warn ( "[KFP] Pipeline version spec 조회 실패, parameters 필터 없이 전달 (mlflow_experiment_name 제외): {}" , e . getMessage ( ) ) ;
return null ;
}
}
/ * *
* Run Request Body 생 성
* KFP Run 생 성 API 요 청 body 구 성 .
* v2beta1 Run : experiment_id , display_name ( 필 수 ) , runtime_config ( 필 수 ) , pipeline_version_reference 등 .
* parameters 는 파 이 프 라 인 스 펙 에 정 의 된 입 력 만 포 함 한 다 . allowedParamNames 가 null 이 면 스 펙 조 회 실 패 로 , mlflow_experiment_name 은 넣 지 않 고 그 외 파 라 미 터 만 전 달 한 다 .
* /
private Map < String , Object > buildKfpRunRequestBody (
CreateRunRequest runRequest ,
Set < String > allowedParamNames
) {
Map < String , Object > body =
new HashMap < > ( ) ;
if ( runRequest . getExperiment_id ( ) ! = null & &
! runRequest . getExperiment_id ( ) . isBlank ( ) ) {
body . put (
"experiment_id" ,
runRequest . getExperiment_id ( )
) ;
private Map < String , Object > buildKfpRunRequestBody ( CreateRunRequest runRequest , Set < String > allowedParamNames ) {
Map < String , Object > body = new HashMap < > ( ) ;
if ( runRequest . getExperiment_id ( ) ! = null & & ! runRequest . getExperiment_id ( ) . isBlank ( ) ) {
body . put ( "experiment_id" , runRequest . getExperiment_id ( ) ) ;
}
body . put (
"display_name" ,
runRequest . getDisplay_name ( ) ! = null
& & ! runRequest . getDisplay_name ( ) . isBlank ( )
? runRequest . getDisplay_name ( )
: "Run"
) ;
if ( runRequest . getDescription ( ) ! = null & &
! runRequest . getDescription ( ) . isBlank ( ) ) {
body . put (
"description" ,
runRequest . getDescription ( )
) ;
// display_name 필수 (KFP v2beta1)
body . put ( "display_name" , runRequest . getDisplay_name ( ) ! = null & & ! runRequest . getDisplay_name ( ) . isBlank ( )
? runRequest . getDisplay_name ( )
: "Run" ) ;
if ( runRequest . getDescription ( ) ! = null & & ! runRequest . getDescription ( ) . isBlank ( ) ) {
body . put ( "description" , runRequest . getDescription ( ) ) ;
}
if ( runRequest . getPipeline_version_reference ( )
! = null ) {
body . put (
"pipeline_version_reference" ,
runRequest . getPipeline_version_reference ( )
) ;
if ( runRequest . getPipeline_version_reference ( ) ! = null ) {
body . put ( "pipeline_version_reference" , runRequest . getPipeline_version_reference ( ) ) ;
}
Map < String , Object > runtimeConfig =
new HashMap < > ( ) ;
if ( runRequest . getRuntime_config ( ) ! = null
& & runRequest . getRuntime_config ( )
. getParameters ( ) ! = null ) {
Map < String , Object > filtered =
new HashMap < > ( ) ;
for ( Map . Entry < String , Object > entry :
runRequest . getRuntime_config ( )
. getParameters ( )
. entrySet ( ) ) {
String key = entry . getKey ( ) ;
if ( allowedParamNames ! = null & &
! allowedParamNames . contains ( key ) ) {
continue ;
if ( runRequest . getService_account ( ) ! = null & & ! runRequest . getService_account ( ) . isBlank ( ) ) {
body . put ( "service_account" , runRequest . getService_account ( ) ) ;
}
// runtime_config 필수 (KFP v2beta1). 파이프라인에 정의된 파라미터만 전달. 정의되지 않으면 mlflow_experiment_name은 넣지 않음.
Map < String , Object > runtimeConfig = new HashMap < > ( ) ;
if ( runRequest . getRuntime_config ( ) ! = null & & runRequest . getRuntime_config ( ) . getParameters ( ) ! = null
& & ! runRequest . getRuntime_config ( ) . getParameters ( ) . isEmpty ( ) ) {
Map < String , Object > params = runRequest . getRuntime_config ( ) . getParameters ( ) ;
Map < String , Object > kfpParams = new HashMap < > ( ) ;
for ( Map . Entry < String , Object > e : params . entrySet ( ) ) {
String key = e . getKey ( ) ;
if ( key = = null ) continue ;
if ( allowedParamNames ! = null ) {
if ( ! allowedParamNames . contains ( key ) ) continue ;
} else {
// 스펙 조회 실패 시: mlflow_* 파라미터는 파이프라인에 없을 수 있으므로 제외
if ( key . startsWith ( "mlflow_" ) ) continue ;
}
filtered . put (
key ,
entry . getValue ( )
) ;
kfpParams . put ( key , e . getValue ( ) ) ;
}
if ( ! filtered . isEmpty ( ) ) {
runtimeConfig . put (
"parameters" ,
filtered
) ;
if ( ! kfpParams . isEmpty ( ) ) {
runtimeConfig . put ( "parameters" , kfpParams ) ;
}
}
body . put (
"runtime_config" ,
runtimeConfig
) ;
body . put ( "runtime_config" , runtimeConfig ) ;
return body ;
}
/ * *
* Experiments 조 회
* /
public Map listExperiments (
String namespace ,
int pageSize ,
String pageToken
) {
public Map listExperiments ( String namespace , int pageSize , String pageToken ) {
try {
UriComponentsBuilder builder = UriComponentsBuilder
. fromHttpUrl ( kubeflowBaseUrl . replaceAll ( "/+$" , "" ) + "/apis/v2beta1/experiments" ) ;
URI uri = UriComponentsBuilder
. fromHttpUrl (
normalizeBaseUrl (
kubeflowBaseUrl
)
)
. path ( "/apis/v2beta1/experiments" )
. queryParamIfPresent (
"namespace" ,
optional ( namespace )
)
. queryParamIfPresent (
"page_token" ,
optional ( pageToken )
)
. queryParam (
"page_size" ,
pageSize
)
. build ( true )
. toUri ( ) ;
if ( namespace ! = null & & ! namespace . isBlank ( ) ) {
builder . queryParam ( "namespace" , namespace ) ;
}
if ( pageSize > 0 ) {
builder . queryParam ( "page_size" , pageSize ) ;
}
if ( pageToken ! = null & & ! pageToken . isBlank ( ) ) {
builder . queryParam ( "page_token" , pageToken ) ;
}
return webClient . get ( )
. uri ( uri )
. retrieve ( )
. bodyToMono ( Map . class )
. block ( ) ;
return webClient . get ( )
. uri ( builder . toUriString ( ) )
. accept ( MediaType . APPLICATION_JSON )
. retrieve ( )
. bodyToMono ( Map . class )
. block ( ) ;
} catch ( Exception e ) {
throw new RuntimeException ( "Kubeflow Experiments 조회 실패" , e ) ;
}
}
/ * *
* Experiment 단 건 조 회
* /
public Map getExperimentById (
String experimentId
) {
public Map getExperimentById ( String experimentId ) {
try {
String url = kubeflowBaseUrl . replaceAll ( "/+$" , "" ) + "/apis/v2beta1/experiments/" + experimentId ;
String uri =
normalizeBaseUrl ( kubeflowBaseUrl )
+ "/apis/v2beta1/experiments/"
+ experimentId ;
return webClient . get ( )
. uri ( url )
. accept ( MediaType . APPLICATION_JSON )
. retrieve ( )
. bodyToMono ( Map . class )
. block ( ) ;
return webClient . get ( )
. uri ( uri )
. retrieve ( )
. bodyToMono ( Map . class )
. block ( ) ;
} catch ( Exception e ) {
throw new RuntimeException ( "Kubeflow experiment 조회 실패: " + experimentId , e ) ;
}
}
/ * *
* Run 단 건 조 회
* KFP Run 단 건 조 회 ( v2beta1 GET / apis / v2beta1 / runs / { run_id } ) .
* run_details . task_details [ ] . pod_name 으 로 UI 로 그 와 동 일 한 Pod 지 정 가 능 .
* /
public Map < String , Object > getKfpRunById (
String runId
) {
if ( runId = = null | |
runId . isBlank ( ) ) {
@SuppressWarnings ( "unchecked" )
public Map < String , Object > getKfpRunById ( String runId ) {
if ( runId = = null | | runId . isBlank ( ) ) {
return null ;
}
String uri =
normalizeBaseUrl ( kubeflowBaseUrl )
+ "/apis/v2beta1/runs/"
+ runId . trim ( ) ;
String url = kubeflowBaseUrl . replaceAll ( "/+$" , "" ) + "/apis/v2beta1/runs/" + runId . trim ( ) ;
try {
return webClient . get ( )
. uri ( ur i )
. uri ( url )
. accept ( MediaType . APPLICATION_JSON )
. retrieve ( )
. bodyToMono ( Map . class )
. block ( ) ;
} catch ( WebClientResponseException e ) {
if ( e . getStatusCode ( ) . value ( ) = = 404 ) {
log . debug (
"Run not found. runId={}" ,
runId
) ;
return null ;
log . debug ( "[KFP] Run not found: {}" , runId ) ;
} else {
log . warn ( "[KFP] GetRun {}: {}" , runId , e . getMessage ( ) ) ;
}
throw e ;
return null ;
} catch ( Exception e ) {
log . debug ( "[KFP] GetRun failed {}: {}" , runId , e . getMessage ( ) ) ;
return null ;
}
}
/ * *
* Node Log 조 회
* KFP ml - pipeline 이 클 러 스 터 에 서 읽 는 Pod 로 그 ( KFP UI 와 동 일 한 백 엔 드 경 로 ) .
* { @code GET / apis / v1beta1 / runs / { run_id } / nodes / { node_id } / log }
* < p > v2 Run ID 도 일 부 배 포 에 서 동 작 합 니 다 . 404 / 비 어 있 으 면 null . < / p >
* /
public String getV1beta1RunNodeLog (
String runId ,
String nodeId
) {
if ( runId = = null | |
nodeId = = null ) {
public String getV1beta1RunNodeLog ( String runId , String nodeId ) {
if ( runId = = null | | runId . isBlank ( ) | | nodeId = = null | | nodeId . isBlank ( ) ) {
return null ;
}
String uri =
normalizeBaseUrl ( kubeflowBaseUrl )
+ "/apis/v1beta1/runs/"
+ UriUtils . encodePathSegment (
runId ,
StandardCharsets . UTF_8
)
+ "/nodes/"
+ UriUtils . encodePathSegment (
nodeId ,
StandardCharsets . UTF_8
)
+ "/log" ;
String base = kubeflowBaseUrl . replaceAll ( "/+$" , "" ) ;
String encRun = UriUtils . encodePathSegment ( runId . trim ( ) , StandardCharsets . UTF_8 ) ;
String encNode = UriUtils . encodePathSegment ( nodeId . trim ( ) , StandardCharsets . UTF_8 ) ;
String url = base + "/apis/v1beta1/runs/" + encRun + "/nodes/" + encNode + "/log" ;
try {
return webClient . get ( )
. uri ( ur i )
. accept ( MediaType . TEXT_PLAIN
. uri ( url )
. accept ( MediaType . TEXT_PLAIN , MediaType . APPLICATION_JSON )
. retrieve ( )
. bodyToMono ( String . class )
. timeout ( Duration . ofSeconds ( 120 ) )
. block ( ) ;
} catch ( WebClientResponseException e ) {
int code = e . getStatusCode ( ) . value ( ) ;
if ( code = = 404 | | code = = 400 ) {
log . debug ( "[KFP] v1beta1 node log {} node={}: {}" , runId , nodeId , code ) ;
} else {
log . debug ( "[KFP] v1beta1 node log {} node={}: {}" , runId , nodeId , e . getMessage ( ) ) ;
}
return null ;
} catch ( Exception e ) {
log . debug (
"Node log 조회 실패" ,
e
) ;
log . debug ( "[KFP] v1beta1 node log failed runId={} node={}: {}" , runId , nodeId , e . getMessage ( ) ) ;
return null ;
}
}
/ * *
* Run 삭 제
* /
public void deleteKfpRun (
String runId ,
String experimentId
) {
String uri =
normalizeBaseUrl ( kubeflowBaseUrl )
+ "/apis/v2beta1/runs/"
+ runId ;
if ( experimentId ! = null & &
! experimentId . isBlank ( ) ) {
uri + = "?experiment_id="
+ experimentId ;
}
webClient . delete ( )
. uri ( uri )
. retrieve ( )
. onStatus (
status - >
status . is4xxClientError ( )
| | status . is5xxServerError ( ) ,
response - >
response . bodyToMono ( String . class )
. map ( msg - >
new RuntimeException (
"KFP DeleteRun failed: "
+ msg
)
)
)
. toBodilessEntity ( )
. block ( ) ;
}
/ * *
* Base URL normalize
* /
private String normalizeBaseUrl (
String baseUrl
) {
if ( baseUrl = = null | |
baseUrl . isBlank ( ) ) {
throw new IllegalArgumentException (
"kubeflow.url is empty"
) ;
}
baseUrl = baseUrl . trim ( ) ;
if ( baseUrl . endsWith ( "/" ) ) {
baseUrl =
baseUrl . substring (
0 ,
baseUrl . length ( ) - 1
) ;
}
return baseUrl ;
}
private Optional < String > optional (
String value
) {
if ( value = = null | |
value . isBlank ( ) ) {
return Optional . empty ( ) ;
}
return Optional . of ( value ) ;
}
/ * *
* Multipart Resource
* KFP Run 삭 제 ( v2beta1 DELETE / apis / v2beta1 / runs / { run_id } ) .
* 성 공 ( 2 xx ) 또 는 이 미 없 음 ( 404 ) 일 때 만 정 상 반 환 . 그 외 4 xx / 5 xx 는 예 외 .
* /
private static class MultipartInputStreamFileResource
extends InputStreamResource {
private final String filename ;
public MultipartInputStreamFileResource (
java . io . InputStream inputStream ,
String filename
) {
super ( inputStream ) ;
this . filename = filename ;
public void deleteKfpRun ( String runId , String experimentId ) {
String base = kubeflowBaseUrl . replaceAll ( "/+$" , "" ) ;
String url = base + "/apis/v2beta1/runs/" + runId ;
if ( experimentId ! = null & & ! experimentId . isBlank ( ) ) {
url = url + "?experiment_id=" + experimentId ;
}
@Override
public String getFilename ( ) {
return filename ;
}
@Override
public long contentLength ( ) {
return - 1 ;
try {
webClient . delete ( )
. uri ( url )
. retrieve ( )
. onStatus ( status - > status . value ( ) = = 404 , resp - > Mono . empty ( ) )
. onStatus ( status - > status . is4xxClientError ( ) | | status . is5xxServerError ( ) ,
resp - > resp . bodyToMono ( String . class )
. doOnNext ( msg - > log . warn ( "[KFP] DeleteRun error {}: {}" , resp . statusCode ( ) , msg ) )
. map ( msg - > new RuntimeException ( "KFP DeleteRun failed: " + resp . statusCode ( ) + " " + msg ) ) )
. toBodilessEntity ( )
. block ( ) ;
} catch ( Exception e ) {
if ( e . getCause ( ) instanceof RuntimeException ) {
throw ( RuntimeException ) e . getCause ( ) ;
}
throw new RuntimeException ( "KFP Run 삭제 실패: " + runId , e ) ;
}
}
}
}
