| /* |
| * 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. |
| */ |
| |
| #include <sstream> |
| #include <list> |
| #include "rpc/RpcAuth.h" |
| #include "common/XmlConfig.h" |
| #include "common/SessionConfig.h" |
| |
| #include "ApplicationClient.h" |
| #include "ApplicationMaster.h" |
| #include "ContainerManagement.h" |
| |
| #include "LibYarnClient.h" |
| #include "LibYarnConstants.h" |
| |
| #include "common/Logger.h" |
| |
| using namespace Yarn::Internal; |
| |
| namespace libyarn { |
| |
| LibYarnClient::LibYarnClient(string &user, string &rmHost, string &rmPort, |
| string &schedHost, string &schedPort, string &amHost, int32_t amPort, |
| string &am_tracking_url, int heartbeatInterval) : |
| amUser(user), schedHost(schedHost), schedPort(schedPort), amHost( |
| amHost), amPort(amPort), am_tracking_url(am_tracking_url), heartbeatInterval( |
| heartbeatInterval), response_id(0), clientJobId(""), keepRun( |
| false), needHeartbeatAlive(false) { |
| pthread_mutex_init(&(heartbeatLock), NULL); |
| amrmClient = NULL; |
| appClient = (void*) new ApplicationClient(user, rmHost, rmPort); |
| nmClient = (void*) new ContainerManagement(); |
| } |
| #ifdef MOCKTEST |
| LibYarnClient::LibYarnClient(string &user,string &rmHost, string &rmPort, string &schedHost, |
| string &schedPort, string &amHost, int32_t amPort, |
| string &am_tracking_url, int heartbeatInterval,Mock::TestLibYarnClientStub *stub): |
| amUser(user),schedHost(schedHost), schedPort(schedPort), amHost(amHost), |
| amPort(amPort), am_tracking_url(am_tracking_url), |
| heartbeatInterval(heartbeatInterval),clientJobId(""), |
| keepRun(false), needHeartbeatAlive(false) { |
| pthread_mutex_init( &(heartbeatLock), NULL ); |
| libyarnStub = stub; |
| appClient = (void*) libyarnStub->getApplicationClient(); |
| amrmClient = (void*) libyarnStub->getApplicationMaster(); |
| nmClient = (void*) libyarnStub->getContainerManagement(); |
| } |
| #endif |
| |
| LibYarnClient::~LibYarnClient() { |
| #ifndef MOCKTEST |
| if (keepRun) { |
| // No need to run heart-beat thread now. |
| keepRun = false; |
| void *thrc = NULL; |
| int rc = pthread_join(heartbeatThread, &thrc); |
| if (rc != 0) { |
| LOG(DEBUG1, "LibYarnClient::~LibYarnClient, fail to join heart-beat thread. " |
| "error code %d", rc); |
| } else { |
| LOG(DEBUG1, "LibYarnClient::~LibYarnClient, join heart-beat thread successfully."); |
| } |
| } |
| #endif |
| if (amrmClient != NULL) { |
| delete (ApplicationMaster*) amrmClient; |
| } |
| delete (ApplicationClient*) appClient; |
| delete (ContainerManagement*) nmClient; |
| } |
| |
| string LibYarnClient::getErrorMessage() { |
| return errorMessage; |
| } |
| |
| void LibYarnClient::setErrorMessage(string errorMsg) { |
| errorMessage = errorMsg; |
| } |
| |
| bool LibYarnClient::isJobHealthy() { |
| return keepRun; |
| } |
| |
| list<ResourceRequest>& LibYarnClient::getAskRequests() { |
| return askRequests; |
| } |
| |
| void LibYarnClient::clearAskRequests() { |
| LOG(DEBUG1, "LibYarnClient::clear ask requests."); |
| askRequests.clear(); |
| } |
| |
| void* heartbeatFunc(void* args) { |
| int failcounter = 0; |
| int retry = 2; |
| LibYarnClient *client = (LibYarnClient*) args; |
| |
| while (client->keepRun) { |
| try { |
| client->dummyAllocate(); |
| failcounter = 0; |
| } catch (const ApplicationMasterNotRegisteredException &e) { |
| /* |
| * In case catch this exception, |
| * heartbeat thread should exits, and re-register AM. |
| */ |
| LOG(WARNING, "LibYarnClient::heartbeat dummy allocation " |
| "catch ApplicationMasterNotRegisteredException. %s", |
| e.msg()); |
| client->keepRun = false; |
| break; |
| } catch (const YarnException &e) { |
| LOG(WARNING, "LibYarnClient::heartbeat dummy allocation " |
| "is not correctly executed with exception raised. %s", |
| e.msg()); |
| failcounter++; |
| if (failcounter > retry) { |
| /* In case retry too many times with errors/exceptions, this |
| * thread will return. LibYarn has to re-register application |
| * and start the heartbeat thread again. |
| */ |
| LOG(WARNING, "LibYarnClient::heartbeatFunc, there are too many " |
| "failures raised. This heart-beat thread exits now."); |
| client->keepRun = false; |
| break; |
| } |
| } |
| usleep((client->heartbeatInterval) * 1000); |
| } |
| |
| LOG(INFO, "LibYarnClient::heartbeatFunc, goes into exit phase."); |
| return (void *) 0; |
| } |
| |
| int LibYarnClient::createJob(string &jobName, string &queue, string &jobId) { |
| try { |
| //Only one jobId for the client right now. |
| if (clientJobId != ""){ |
| throw std::invalid_argument( "Exist an application for the client"); |
| } |
| ApplicationClient *applicationClient = (ApplicationClient*) appClient; |
| |
| //1. getNewApplication |
| ApplicationId appId = applicationClient->getNewApplication(); |
| LOG(DEBUG1, "LibYarnClient::createJob, getNewApplication finished, appId:[clusterTimeStamp:%lld,id:%d]", |
| appId.getClusterTimestamp(), appId.getId()); |
| |
| //2. submitApplication |
| ApplicationSubmissionContext appSubmitCtx; |
| appSubmitCtx.setApplicationId(appId); |
| appSubmitCtx.setApplicationName(jobName); |
| appSubmitCtx.setQueue(queue); |
| |
| Priority priority(0); |
| appSubmitCtx.setPriority(priority); |
| |
| ContainerLaunchContext amContainerSpec; |
| appSubmitCtx.setAMContainerSpec(amContainerSpec); |
| |
| appSubmitCtx.setUnmanagedAM(true); |
| appSubmitCtx.setMaxAppAttempts(1); |
| |
| applicationClient->submitApplication(appSubmitCtx); |
| LOG(DEBUG1, "LibYarnClient::createJob, submitApplication finished"); |
| |
| //3. wait util AM is ACCEPTED and return the AMRMToken |
| ApplicationReport report; |
| int retry = 10; |
| while (retry > 0) { |
| report = applicationClient->getApplicationReport(appId); |
| LOG(DEBUG1, "LibYarnClient::createJob, appId[cluster_timestamp:%lld,id:%d], appState:%d", |
| appId.getClusterTimestamp(), appId.getId(), report.getYarnApplicationState()); |
| if ((report.getAMRMToken().getPassword() != "") && |
| report.getYarnApplicationState() == YarnApplicationState::ACCEPTED) { |
| break; |
| } else { |
| retry--; |
| usleep(1000000L); |
| } |
| if (retry == 0) |
| THROW(AccessControlException, "can not register application to YARN"); |
| } |
| |
| clientAppId = appId; |
| clientAppAttempId = report.getCurrentAppAttemptId(); |
| |
| //4.1 new ApplicationMaster |
| Token token = report.getAMRMToken(); |
| UserInfo user; |
| if (applicationClient->getMethod() == SIMPLE) { |
| user = UserInfo::LocalUser(); |
| } else if (applicationClient->getMethod() == KERBEROS) { |
| user.setEffectiveUser(applicationClient->getPrincipal()); |
| user.setRealUser(applicationClient->getUser()); |
| } else { |
| LOG(WARNING, "LibYarnClient::createJob: unsupported RPC method:%d. ", |
| applicationClient->getMethod()); |
| } |
| |
| Yarn::Token AMToken; |
| AMToken.setIdentifier(token.getIdentifier()); |
| AMToken.setKind(token.getKind()); |
| AMToken.setPassword(token.getPassword()); |
| AMToken.setService(token.getService()); |
| |
| user.addToken(AMToken); |
| #ifndef MOCKTEST |
| amrmClient = (void*) new ApplicationMaster(this->schedHost, this->schedPort, |
| user, token.getService()); |
| #endif |
| |
| //4.2 register to RM scheduler as AM |
| ((ApplicationMaster*) amrmClient)->registerApplicationMaster(amHost, amPort, am_tracking_url); |
| LOG(DEBUG1, "LibYarnClient::createJob, registerApplicationMaster finished"); |
| |
| #ifndef MOCKTEST |
| keepRun = true; |
| //5. setup the heartbeat thread to allocate, release, heartbeat |
| int rc = pthread_create(&heartbeatThread, NULL, heartbeatFunc, this); |
| if (rc != 0) { |
| keepRun = false; |
| LOG(WARNING, "LibYarnClient::createJob, fail to create heart-beat thread. " |
| "error code %d", rc); |
| throw std::runtime_error("Fail to create heart-beat thread."); |
| } |
| needHeartbeatAlive = true; |
| #endif |
| |
| LOG(DEBUG1, "LibYarnClient::createJob, after AM register to RM, a heartbeat thread has been started"); |
| //6. return jobId |
| std::stringstream ss; |
| ss << "job_" << appId.getClusterTimestamp() << "_" << appId.getId(); |
| jobId = ss.str(); |
| |
| LOG(INFO, "LibYarnClient::createJob, appId[cluster_timestamp:%lld,id:%d]", |
| clientAppId.getClusterTimestamp(), clientAppId.getId()); |
| clientJobId = jobId; |
| return FR_SUCCEEDED; |
| } catch (const YarnNetworkConnectException &e) { |
| std::stringstream errorMsg; |
| errorMsg << "LibYarnClient::createJob, catch network connection exception:" << e.what(); |
| setErrorMessage(errorMsg.str()); |
| return FR_FAILED; |
| } catch (const std::exception &e) { |
| std::stringstream errorMsg; |
| errorMsg << "LibYarnClient::createJob, catch exception:" << e.what(); |
| setErrorMessage(errorMsg.str()); |
| return FR_FAILED; |
| } catch (...) { |
| std::stringstream errorMsg; |
| errorMsg << "LibYarnClient::createJob, catch unexpected exception."; |
| setErrorMessage(errorMsg.str()); |
| return FR_FAILED; |
| } |
| } |
| |
| int LibYarnClient::forceKillJob(string &jobId) { |
| |
| #ifndef MOCKTEST |
| if (keepRun) { |
| keepRun = false; |
| void *thrc = NULL; |
| int rc = pthread_join(heartbeatThread, &thrc); |
| if (rc != 0) { |
| LOG(WARNING, "LibYarnClient::forceKillJob, fail to join heart-beat thread. " |
| "error code %d", rc); |
| return FR_FAILED; |
| } else { |
| LOG(INFO, "LibYarnClient::forceKillJob, join heart-beat thread successfully."); |
| } |
| } |
| #endif |
| |
| try{ |
| if (jobId != clientJobId) { |
| throw std::invalid_argument("The jobId is wrong, please check the jobId argument"); |
| } |
| |
| needHeartbeatAlive = false; |
| for (map<int64_t, Container*>::iterator it = jobIdContainers.begin(); |
| it != jobIdContainers.end(); it++) { |
| std::ostringstream key; |
| Container *container = it->second; |
| key << container->getNodeId().getHost() << ":" << container->getNodeId().getPort(); |
| Token nmToken = nmTokenCache[key.str()]; |
| ((ContainerManagement*) nmClient)->stopContainer((*container), nmToken); |
| LOG(DEBUG1, "LibYarnClient::forceKillJob, container:%ld is stopped",container->getId().getId()); |
| } |
| |
| ((ApplicationClient*) appClient)->forceKillApplication(clientAppId); |
| LOG(INFO, "LibYarnClient::force to kill this application."); |
| |
| for (map<int64_t,Container*>::iterator it = jobIdContainers.begin(); |
| it != jobIdContainers.end(); it++) { |
| LOG(DEBUG1, "LibYarnClient::forceKillJob, container:%ld in jobIdContainers is deleted", |
| it->second->getId().getId()); |
| delete it->second; |
| it->second = NULL; |
| } |
| jobIdContainers.clear(); |
| activeFailContainerIds.clear(); |
| return FR_SUCCEEDED; |
| } catch (std::exception& e) { |
| std::stringstream errorMsg; |
| errorMsg << "LibYarnClient::forceKillJob, catch the exception:" << e.what(); |
| setErrorMessage(errorMsg.str()); |
| return FR_FAILED; |
| } catch (...) { |
| std::stringstream errorMsg; |
| errorMsg << "LibYarnClient::forceKillJob, catch unexpected exception."; |
| setErrorMessage(errorMsg.str()); |
| return FR_FAILED; |
| } |
| } |
| |
| |
| void LibYarnClient::dummyAllocate() { |
| |
| pthread_mutex_lock(&heartbeatLock); |
| ApplicationMaster* amrmClientAlias = (ApplicationMaster*) amrmClient; |
| |
| //1) requestProto_blank |
| list<ResourceRequest> asksBlank; |
| //2) releasesBlank |
| list<ContainerId> releasesBlank; |
| //3) blacklistRequestBlank |
| ResourceBlacklistRequest blacklistRequestBlank; |
| //4) progress |
| float progress = 0.5; |
| int allocatedNum = 0; |
| |
| try { |
| LOG(DEBUG1, "LibYarnClient::dummyAllocate, do a AM-RM heartbeat with response_id:%d", response_id); |
| AllocateResponse response = amrmClientAlias->allocate(asksBlank, |
| releasesBlank, blacklistRequestBlank, response_id, progress); |
| response_id = response.getResponseId(); |
| list<Container> allocatedContainers = response.getAllocatedContainers(); |
| allocatedNum = allocatedContainers.size(); |
| LOG(DEBUG1, "LibYarnClient::dummyAllocate returned response_id :%d", response_id); |
| if (allocatedNum > 0) { |
| /* |
| * In rare case, client gets allocated containers in heartbeat, |
| * free them immediately. |
| */ |
| LOG(DEBUG1, "LibYarnClient::dummyAllocate returned allocated size: %d, " |
| "free them immediately.", allocatedNum); |
| list<ContainerId> releases; |
| for (list<Container>::iterator it = allocatedContainers.begin(); |
| it != allocatedContainers.end(); it++) { |
| releases.push_back((*it).getId()); |
| } |
| list<ResourceRequest> asksBlank; |
| ResourceBlacklistRequest blacklistRequestBlank; |
| response = amrmClientAlias->allocate(asksBlank, releases, |
| blacklistRequestBlank, response_id, progress); |
| response_id = response.getResponseId(); |
| } |
| pthread_mutex_unlock(&heartbeatLock); |
| } |
| catch (const YarnException &e) { |
| LOG(WARNING, "LibYarnClient::dummyAllocate, dummy allocation " |
| "is not correctly executed with exception raised. %s", |
| e.msg()); |
| pthread_mutex_unlock(&heartbeatLock); |
| throw; |
| } |
| } |
| |
| void LibYarnClient::addResourceRequest(Resource capability, |
| int32_t num_containers, string host, int32_t priority, |
| bool relax_locality) |
| { |
| ResourceRequest *req = new ResourceRequest(); |
| req->setCapability(capability); |
| req->setNumContainers(num_containers); |
| Priority priorityProto; |
| priorityProto.setPriority(priority); |
| req->setPriority(priorityProto); |
| req->setResourceName(host); |
| req->setRelaxLocality(relax_locality); |
| try { |
| askRequests.push_back(*req); |
| LOG(DEBUG1, "LibYarnClient::put a request into ask list, " |
| "mem:%d, cpu:%d, priority:%d, resource name:%s, relax:%d, num_containers:%d", |
| capability.getMemory(), capability.getVirtualCores(), priority, host.c_str(), |
| relax_locality, num_containers); |
| } catch (std::exception &e) { |
| LOG(WARNING, "LibYarnClient::Fail to add a resource request " |
| "to ask list. %s ", e.what()); |
| } |
| } |
| |
| /* |
| * This function creates resource requests according to the requirement of the caller, then put these |
| * requests into a list. The logic is a little similar to java client AMRMClient:addContainerRequest(). |
| * There are three level nodes in YARN: off-switch(aka ANY), rack, host.By now, only ANY-level and host-level |
| * is supported. |
| * |
| * Parameters: |
| * jobId: jobId |
| * capability: the quota of the resource |
| * count: the required number of the containers |
| * preferred: node list, NULL means ANY host. If one node's rack name is NULL, a default rack name is set. |
| * priority: priority |
| */ |
| int LibYarnClient::addContainerRequests(string &jobId, Resource &capability, |
| int32_t num_containers, list<LibYarnNodeInfo> &preferred, |
| int32_t priority, bool relax_locality) |
| { |
| try { |
| if (jobId != clientJobId) { |
| throw std::invalid_argument("The jobId is wrong, check the jobId argument"); |
| } |
| |
| map<string, int32_t> inferredRacks; |
| |
| for (list<LibYarnNodeInfo>::iterator iter = preferred.begin(); |
| iter != preferred.end(); iter++) { |
| LOG(DEBUG1, "LibYarnClient::addContainerRequests, " |
| "get a preferred host info, host:%s,rack:%s,container number:%d", |
| iter->getHost().c_str(), iter->getRack().c_str(), iter->getContainerNum()); |
| /* add a resource request for this node */ |
| addResourceRequest(capability, iter->getContainerNum(), iter->getHost(), priority, true); |
| map<string, int32_t>::iterator it = inferredRacks.find(iter->getRack()); |
| if (it != inferredRacks.end()) |
| it->second += iter->getContainerNum(); |
| else |
| inferredRacks.insert(make_pair(iter->getRack(), iter->getContainerNum())); |
| } |
| |
| /* add resource requests for racks*/ |
| for (map<string, int32_t>::iterator it = inferredRacks.begin(); |
| it != inferredRacks.end(); it++) { |
| addResourceRequest(capability, it->second, it->first, priority, relax_locality); |
| } |
| |
| /* add resource request for off-switch */ |
| addResourceRequest(capability, num_containers, YARN_HOST_ANY, priority, relax_locality); |
| |
| return FR_SUCCEEDED; |
| } catch (std::exception &e) { |
| std::stringstream errorMsg; |
| errorMsg << "LibYarnClient::addContainerRequests catch std exception:" << e.what(); |
| setErrorMessage(errorMsg.str()); |
| return FR_FAILED; |
| } catch (...) { |
| std::stringstream errorMsg; |
| errorMsg << "LibYarnClient::addContainerRequests catch unexpected exception."; |
| setErrorMessage(errorMsg.str()); |
| return FR_FAILED; |
| } |
| } |
| |
| /* |
| message AllocateRequestProto { |
| repeated ResourceRequestProto ask = 1; |
| repeated ContainerIdProto release = 2; |
| optional ResourceBlacklistRequestProto blacklist_request = 3; |
| optional int32 response_id = 4; |
| optional float progress = 5; |
| } |
| |
| message AllocateResponseProto { |
| optional AMCommandProto a_m_command = 1; |
| optional int32 response_id = 2; |
| repeated ContainerProto allocated_containers = 3; |
| repeated ContainerStatusProto completed_container_statuses = 4; |
| optional ResourceProto limit = 5; |
| repeated NodeReportProto updated_nodes = 6; |
| optional int32 num_cluster_nodes = 7; |
| optional PreemptionMessageProto preempt = 8; |
| repeated NMTokenProto nm_tokens = 9; |
| } |
| */ |
| int LibYarnClient::allocateResources(string &jobId, |
| list<string> &blackListAdditions, list<string> &blackListRemovals, |
| list<Container> &allocatedContainers, int32_t num_containers) |
| { |
| try{ |
| AllocateResponse response; |
| int retry = 5; |
| int allocatedNumOnce = 0; |
| int allocatedNumTotal = 0; |
| |
| pthread_mutex_lock(&heartbeatLock); |
| if (jobId != clientJobId) { |
| throw std::invalid_argument("The jobId is wrong, check the jobId argument"); |
| } |
| |
| if (!keepRun && needHeartbeatAlive) { |
| throw std::runtime_error("LibYarnClient::libyarn AM heartbeat thread has stopped."); |
| } |
| |
| ApplicationMaster* amrmClientAlias = (ApplicationMaster*) amrmClient; |
| list<Container> allocatedContainerCache; |
| list<ContainerReport> preContainerReports; |
| preContainerReports = ((ApplicationClient*) appClient)->getContainers(clientAppAttempId); |
| |
| list<ContainerId> releasesBlank; |
| ResourceBlacklistRequest blacklistRequest; |
| blacklistRequest.setBlacklistAdditions(blackListAdditions); |
| blacklistRequest.setBlacklistRemovals(blackListRemovals); |
| float progress = 0.5; |
| |
| LOG(DEBUG1, "LibYarnClient::request to allocate resource, container number:%d", |
| num_containers); |
| |
| while (retry > 0) { |
| LOG(DEBUG1, "LibYarnClient::allocate with response id : %d", response_id); |
| AllocateResponse response = amrmClientAlias->allocate( |
| this->askRequests, releasesBlank, blacklistRequest, |
| response_id, progress); |
| response_id = response.getResponseId(); |
| LOG(DEBUG1, "LibYarnClient::allocate returned response id : %d", response_id); |
| list<NMToken> nmTokens = response.getNMTokens(); |
| for (list<NMToken>::iterator it = nmTokens.begin(); it != nmTokens.end(); it++) { |
| std::ostringstream oss; |
| oss << (*it).getNodeId().getHost() << ":" << (*it).getNodeId().getPort(); |
| nmTokenCache[oss.str()] = (*it).getToken(); |
| } |
| this->clearAskRequests(); |
| list<Container> allocatedContainerOnce = response.getAllocatedContainers(); |
| allocatedNumOnce = allocatedContainerOnce.size(); |
| if (allocatedNumOnce <= 0) { |
| LOG(DEBUG1, "LibYarnClient:: fail to allocate from YARN RM, try again"); |
| retry--; |
| if(retry == 0 && allocatedNumTotal == 0) { |
| /* If failed, just return to Resource Broker to handle*/ |
| pthread_mutex_unlock(&heartbeatLock); |
| LOG(WARNING,"LibYarnClient:: fail to allocate from YARN RM after retry several times"); |
| return FR_SUCCEEDED; |
| } |
| } else { |
| allocatedNumTotal += allocatedNumOnce; |
| allocatedContainerCache.insert(allocatedContainerCache.end(), allocatedContainerOnce.begin(), allocatedContainerOnce.end()); |
| LOG(DEBUG1, "LibYarnClient:: allocate %d containers from YARN RM", allocatedNumOnce); |
| if (allocatedNumTotal >= num_containers) { |
| LOG(DEBUG1, "LibYarnClient:: allocate enough containers from YARN RM, " |
| "expected:%d, total:%d", num_containers, allocatedNumTotal); |
| break; |
| } |
| |
| } |
| usleep(TimeInterval::ALLOCATE_INTERVAL_MS); |
| } |
| |
| LOG(INFO,"LibYarnClient::allocate resource, response_id:%d, allocated container number:%d", |
| response_id, allocatedNumTotal); |
| |
| /* a workaround for allocate more container than request */ |
| list<ContainerId> releases; |
| list<ContainerReport> afterContainerReports; |
| afterContainerReports = ((ApplicationClient*) appClient)->getContainers(clientAppAttempId); |
| for (list<ContainerReport>::iterator ait = afterContainerReports.begin(); |
| ait != afterContainerReports.end(); ait++) { |
| bool foundInPre = false; |
| for (list<ContainerReport>::iterator pit = |
| preContainerReports.begin(); |
| pit != preContainerReports.end(); pit++) { |
| if (pit->getId().getId() == ait->getId().getId()) { |
| foundInPre = true; |
| break; |
| } |
| } |
| if (!foundInPre) { |
| bool foundInNewAllocated = false; |
| for (list<Container>::iterator cit = |
| allocatedContainerCache.begin(); |
| cit != allocatedContainerCache.end(); cit++) { |
| if (cit->getId().getId() == ait->getId().getId()) { |
| foundInNewAllocated = true; |
| break; |
| } |
| } |
| if (!foundInNewAllocated) { |
| releases.push_back((*ait).getId()); |
| } |
| } |
| } |
| |
| int totalNeedRelease = allocatedContainerCache.size() - num_containers; |
| LOG(DEBUG1, "LibYarnClient::allocateResources, total_allocated_containers:%ld, total_need_release:%d", |
| allocatedContainerCache.size(), totalNeedRelease); |
| if(totalNeedRelease > 0) { |
| for (int i = 0; i < totalNeedRelease; i++) { |
| list<Container>::iterator it = allocatedContainerCache.begin(); |
| releases.push_back((*it).getId()); |
| allocatedContainerCache.erase(it); |
| } |
| |
| list<ResourceRequest> asksBlank; |
| ResourceBlacklistRequest blacklistRequestBlank; |
| response = amrmClientAlias->allocate(asksBlank, releases, |
| blacklistRequestBlank, response_id, progress); |
| response_id = response.getResponseId(); |
| } |
| |
| /* 3. store allocated containers */ |
| for (list<Container>::iterator it = allocatedContainerCache.begin(); |
| it != allocatedContainerCache.end(); it++) { |
| Container *container = new Container((*it)); |
| int64_t containerId = container->getId().getId(); |
| jobIdContainers[containerId] = container; |
| } |
| allocatedContainers = allocatedContainerCache; |
| |
| LOG(DEBUG1, "LibYarnClient::allocateResources, put all allocated containers size:%ld", |
| allocatedContainerCache.size()); |
| |
| pthread_mutex_unlock(&heartbeatLock); |
| |
| return FR_SUCCEEDED; |
| } catch(std::exception &e) { |
| std::stringstream errorMsg; |
| |
| errorMsg << "LibYarnClient::allocateResources, catch exception:" << e.what(); |
| setErrorMessage(errorMsg.str()); |
| |
| pthread_mutex_unlock(&heartbeatLock); |
| return FR_FAILED; |
| } catch (const ApplicationMasterNotRegisteredException &e) { |
| std::stringstream errorMsg; |
| |
| errorMsg << "LibYarnClient::allocateResources, " |
| "catch ApplicationMasterNotRegisteredException." << e.what(); |
| setErrorMessage(errorMsg.str()); |
| |
| pthread_mutex_unlock(&heartbeatLock); |
| return FR_FAILED; |
| } catch (...) { |
| std::stringstream errorMsg; |
| |
| errorMsg << "LibYarnClient::allocateResources, catch unexpected exception."; |
| setErrorMessage(errorMsg.str()); |
| |
| pthread_mutex_unlock(&heartbeatLock); |
| return FR_FAILED; |
| } |
| } |
| |
| int LibYarnClient::releaseResources(string &jobId,int64_t releaseContainerIds[], int releaseContainerSize) |
| { |
| try{ |
| pthread_mutex_lock(&heartbeatLock); |
| if (jobId != clientJobId) { |
| throw std::invalid_argument("The jobId is wrong,please check the jobId argument"); |
| } |
| |
| if (!keepRun && needHeartbeatAlive) { |
| throw std::runtime_error("LibYarnClient::libyarn AM heartbeat thread has stopped."); |
| } |
| |
| ApplicationMaster* amrmClientAlias = (ApplicationMaster*) amrmClient; |
| //1) asksBlank |
| list<ResourceRequest> asksBlank; |
| //2) releases |
| list<ContainerId> releases; |
| for (int i = 0; i < releaseContainerSize; i++) { |
| int64_t containerId = releaseContainerIds[i]; |
| map<int64_t, Container*>::iterator it = jobIdContainers.find(containerId); |
| if (it != jobIdContainers.end()) { |
| releases.push_back((it->second)->getId()); |
| } |
| } |
| //3) blacklistRequestBlank |
| ResourceBlacklistRequest blacklistRequestBlank; |
| //4) progress |
| float progress = 0.5; |
| |
| AllocateResponse response = amrmClientAlias->allocate(asksBlank, |
| releases, blacklistRequestBlank, response_id, progress); |
| response_id = response.getResponseId(); |
| //erase from the map jobIdContainers |
| for (list<ContainerId>::iterator it = releases.begin(); it != releases.end(); it++) { |
| LOG(DEBUG1, "LibYarnClient::releaseResource, released ContainerId:%ld", |
| it->getId()); |
| map<int64_t, Container*>::iterator cit = jobIdContainers.find(it->getId()); |
| if (cit != jobIdContainers.end()) { |
| delete cit->second; |
| cit->second = NULL; |
| jobIdContainers.erase(it->getId()); |
| } |
| //erase the element if in activeFailContainers |
| set<int64_t>::iterator sit = activeFailContainerIds.find(it->getId()); |
| if (sit != activeFailContainerIds.end()) { |
| LOG(INFO, "LibYarnClient::releaseResource, remove %ld from activeFailContainerIds", |
| (*sit)); |
| activeFailContainerIds.erase(*sit); |
| } |
| } |
| LOG(INFO, "LibYarnClient::release resources, container number:%d", |
| releases.size()); |
| pthread_mutex_unlock(&heartbeatLock); |
| return FR_SUCCEEDED; |
| } catch (std::exception &e) { |
| std::stringstream errorMsg; |
| errorMsg << "LibYarnClient::releaseResources, catch exception:" << e.what(); |
| setErrorMessage(errorMsg.str()); |
| pthread_mutex_unlock(&heartbeatLock); |
| return FR_FAILED; |
| } catch (...) { |
| std::stringstream errorMsg; |
| errorMsg << "LibYarnClient::releaseResources, catch unexpected exception."; |
| setErrorMessage(errorMsg.str()); |
| pthread_mutex_unlock(&heartbeatLock); |
| return FR_FAILED; |
| } |
| } |
| |
| /* |
| --------------------- |
| message StartContainerRequestProto { |
| optional ContainerLaunchContextProto container_launch_context = 1; |
| optional hadoop.common.TokenProto container_token = 2; |
| } |
| |
| message StartContainerResponseProto { |
| repeated StringBytesMapProto services_meta_data = 1; |
| } |
| --------------------- |
| rpc startContainers(StartContainersRequestProto) returns (StartContainersResponseProto); |
| --------------------- |
| message StartContainersRequestProto { |
| repeated StartContainerRequestProto start_container_request = 1; |
| } |
| |
| message StartContainersResponseProto { |
| repeated StringBytesMapProto services_meta_data = 1; |
| repeated ContainerIdProto succeeded_requests = 2; |
| repeated ContainerExceptionMapProto failed_requests = 3; |
| } |
| */ |
| int LibYarnClient::activeResources(string &jobId,int64_t activeContainerIds[],int activeContainerSize) { |
| try{ |
| if (jobId != clientJobId) { |
| throw std::invalid_argument("The jobId is wrong,please check the jobId argument"); |
| } |
| if (!keepRun && needHeartbeatAlive) { |
| throw std::runtime_error("LibYarnClient::libyarn AM heartbeat thread has stopped."); |
| } |
| |
| LOG(DEBUG1, "LibYarnClient::activeResources, activeResources started"); |
| |
| for (int i = 0; i < activeContainerSize; i++) { |
| int64_t containerId = activeContainerIds[i]; |
| map<int64_t, Container*>::iterator it = jobIdContainers.find(containerId); |
| if (it != jobIdContainers.end()) { |
| try { |
| Container *container = it->second; |
| std::ostringstream key; |
| key << container->getNodeId().getHost() << ":" << container->getNodeId().getPort(); |
| |
| Token nmToken = nmTokenCache[key.str()]; |
| string cmd("sleep 10000000000"); |
| list<string> cmds; |
| cmds.push_back(cmd); |
| ContainerLaunchContext ctx; |
| ctx.setCommand(cmds); |
| StartContainerRequest request; |
| request.setContainerLaunchCtx(ctx); |
| Token cToken = container->getContainerToken(); |
| request.setContainerToken(cToken); |
| LOG(DEBUG1, "LibYarnClient::activeResources active containerId:%ld", containerId); |
| ((ContainerManagement*)nmClient)->startContainer((*container), request, nmToken); |
| } catch (std::exception& e) { |
| LOG(WARNING, "LibYarnClient::activeResources, activeResources Failed Id:%ld,exception:%s", |
| containerId,e.what()); |
| activeFailContainerIds.insert(containerId); |
| } |
| } |
| } |
| |
| LOG(INFO, "LibYarnClient::active resources, container number:%d", |
| activeContainerSize); |
| return FR_SUCCEEDED; |
| } catch (std::exception& e) { |
| std::stringstream errorMsg; |
| errorMsg << "LibYarnClient::activeResources, Catch the Exception:" << e.what(); |
| setErrorMessage(errorMsg.str()); |
| return FR_FAILED; |
| } catch (...) { |
| std::stringstream errorMsg; |
| errorMsg << "LibYarnClient::activeResources, catch unexpected exception."; |
| setErrorMessage(errorMsg.str()); |
| return FR_FAILED; |
| } |
| } |
| int LibYarnClient::getActiveFailContainerIds(set<int64_t> &activeFailIds){ |
| activeFailIds = activeFailContainerIds; |
| return FR_SUCCEEDED; |
| } |
| |
| int LibYarnClient::finishJob(string &jobId, FinalApplicationStatus finalStatus) { |
| |
| #ifndef MOCKTEST |
| if (keepRun) { |
| // No need to run heart-beat thread now. |
| keepRun = false; |
| void *thrc = NULL; |
| int rc = pthread_join(heartbeatThread, &thrc); |
| if (rc != 0) { |
| LOG(WARNING, "LibYarnClient::finishJob, fail to join heart-beat thread. " |
| "error code %d", rc); |
| return FR_FAILED; |
| } else { |
| LOG(INFO, "LibYarnClient::finishJob, join heart-beat thread successfully."); |
| } |
| } |
| #endif |
| |
| try { |
| if (jobId != clientJobId) { |
| throw std::invalid_argument("The jobId is wrong,please check the jobId argument"); |
| } |
| needHeartbeatAlive = false; |
| |
| //1. we should stop all containers related with this job |
| for (map<int64_t, Container*>::iterator it = jobIdContainers.begin(); it != jobIdContainers. end(); it++) { |
| std::ostringstream key; |
| Container *container = it->second; |
| key << container->getNodeId().getHost() << ":" << container->getNodeId().getPort(); |
| Token nmToken = nmTokenCache[key.str()]; |
| ((ContainerManagement*) nmClient)->stopContainer((*container), nmToken); |
| LOG(DEBUG1, "LibYarnClient::finishJob, container:%ld is stopped",container->getId().getId()); |
| } |
| LOG(DEBUG1, "LibYarnClient::finishJob, all containers for jobId:%s are stopped",jobId.c_str()); |
| //2. finish AM |
| string diagnostics(""); |
| string tracking_url(""); |
| ((ApplicationMaster*) amrmClient)->finishApplicationMaster(diagnostics, tracking_url, finalStatus); |
| LOG(INFO, "LibYarnClient::finishJob, finish AM for jobId:%s, finalStatus:%d", jobId.c_str(), finalStatus); |
| //free the Container* memory |
| for (map<int64_t, Container*>::iterator it = jobIdContainers.begin(); it != jobIdContainers.end(); it++) { |
| LOG(DEBUG1, "LibYarnClient::finishJob, container:%ld in jobIdContainers is deleted", |
| it->second->getId().getId()); |
| delete it->second; |
| it->second = NULL; |
| } |
| jobIdContainers.clear(); |
| activeFailContainerIds.clear(); |
| |
| return FR_SUCCEEDED; |
| } catch (std::exception& e) { |
| std::stringstream errorMsg; |
| errorMsg << "LibYarnClient::finishJob, catch the Exception:" << e.what(); |
| setErrorMessage(errorMsg.str()); |
| return FR_FAILED; |
| } catch (const ApplicationMasterNotRegisteredException &e) { |
| std::stringstream errorMsg; |
| errorMsg << "LibYarnClient::finishJob, " |
| "catch ApplicationMasterNotRegisteredException." << e.what(); |
| setErrorMessage(errorMsg.str()); |
| return FR_FAILED; |
| } catch (...) { |
| std::stringstream errorMsg; |
| errorMsg << "LibYarnClient::finishJob, catch unexpected exception."; |
| setErrorMessage(errorMsg.str()); |
| return FR_FAILED; |
| } |
| } |
| |
| int LibYarnClient::getApplicationReport(string &jobId,ApplicationReport &applicationReport) { |
| try { |
| if (jobId != clientJobId) { |
| throw std::invalid_argument("The jobId is wrong,please check the jobId argument"); |
| } |
| |
| if (!keepRun && needHeartbeatAlive) { |
| throw std::runtime_error("LibYarnClient::libyarn AM heartbeat thread has stopped."); |
| } |
| |
| LOG(DEBUG1, "LibYarnClient::getApplicationReport, appId[cluster_timestamp:%lld,id:%d]", |
| clientAppId.getClusterTimestamp(), clientAppId.getId()); |
| applicationReport = ((ApplicationClient*) appClient)->getApplicationReport(clientAppId); |
| LOG(DEBUG1, "LibYarnClient::getApplicationReport, appId[cluster_timestamp:%lld,id:%d],getCurrentAppAttemptId:%d", |
| applicationReport.getApplicationId().getClusterTimestamp(), |
| applicationReport.getApplicationId().getId(), |
| applicationReport.getCurrentAppAttemptId().getAttemptId()); |
| |
| return FR_SUCCEEDED; |
| } catch (std::exception& e) { |
| std::stringstream errorMsg; |
| errorMsg << "LibYarnClient::getApplicationReport, Catch the Exception:" |
| << e.what(); |
| setErrorMessage(errorMsg.str()); |
| return FR_FAILED; |
| } catch (...) { |
| std::stringstream errorMsg; |
| errorMsg << "LibYarnClient::getApplicationReport, catch unexpected exception."; |
| setErrorMessage(errorMsg.str()); |
| return FR_FAILED; |
| } |
| } |
| |
| int LibYarnClient::getContainerReports(string &jobId,list<ContainerReport> &containerReports) { |
| try { |
| if (jobId != clientJobId) { |
| throw std::invalid_argument("The jobId is wrong,please check the jobId argument"); |
| } |
| |
| if (!keepRun && needHeartbeatAlive) { |
| throw std::runtime_error("LibYarnClient::libyarn AM heartbeat thread has stopped."); |
| } |
| |
| LOG(DEBUG1, "LibYarnClient::getContainerReports, appId[cluster_timestamp:%lld,id:%d]", |
| clientAppId.getClusterTimestamp(), clientAppId.getId()); |
| containerReports = ((ApplicationClient*) appClient)->getContainers(clientAppAttempId); |
| return FR_SUCCEEDED; |
| } catch (std::exception& e) { |
| std::stringstream errorMsg; |
| errorMsg << "LibYarnClient::getContainerReports, catch the Exception:" << e.what(); |
| setErrorMessage(errorMsg.str()); |
| return FR_FAILED; |
| } catch (...) { |
| std::stringstream errorMsg; |
| errorMsg << "LibYarnClient::getContainerReports, catch unexpected exception."; |
| setErrorMessage(errorMsg.str()); |
| return FR_FAILED; |
| } |
| } |
| |
| int LibYarnClient::getContainerStatuses(string &jobId, int64_t containerIds[], |
| int containerSize, list<ContainerStatus> &containerStatues) |
| { |
| try { |
| if (jobId != clientJobId) { |
| throw std::invalid_argument("The jobId is wrong,please check the jobId argument"); |
| } |
| |
| if (!keepRun && needHeartbeatAlive) { |
| throw std::runtime_error("LibYarnClient::libyarn AM heartbeat thread has stopped."); |
| } |
| |
| for (int i = 0; i < containerSize; i++) { |
| int64_t containerId = containerIds[i]; |
| map<int64_t, Container*>::iterator it = jobIdContainers.find(containerId); |
| if (it != jobIdContainers.end()) { |
| try { |
| Container *container = it->second; |
| std::ostringstream key; |
| key << container->getNodeId().getHost() << ":"<< container->getNodeId().getPort(); |
| Token nmToken = nmTokenCache[key.str()]; |
| ContainerStatus containerStatus = ((ContainerManagement*) nmClient)->getContainerStatus((*container), nmToken); |
| // the response containerId will be 0 if the request containerId is not exist |
| if (containerStatus.getContainerId().getId() != 0){ |
| containerStatues.push_back(containerStatus); |
| } |
| } catch (std::exception& e) { |
| LOG(INFO, "LibYarnClient::getContainerStatuses, getContainerStatuses Failed Id:%ld,exception:%s", |
| containerId, e.what()); |
| } |
| } |
| } |
| return FR_SUCCEEDED; |
| } catch (std::exception& e) { |
| std::stringstream errorMsg; |
| errorMsg << "LibYarnClient::getContainerStatuses, Catch the Exception:" << e.what(); |
| setErrorMessage(errorMsg.str()); |
| return FR_FAILED; |
| } catch (...) { |
| std::stringstream errorMsg; |
| errorMsg << "LibYarnClient::getContainerStatuses, catch unexpected exception."; |
| setErrorMessage(errorMsg.str()); |
| return FR_FAILED; |
| } |
| } |
| |
| int LibYarnClient::getQueueInfo(string &queue, bool includeApps, |
| bool includeChildQueues, bool recursive, QueueInfo &queueInfo) |
| { |
| try { |
| if (!keepRun && needHeartbeatAlive) { |
| throw std::runtime_error("LibYarnClient::libyarn AM heartbeat thread has stopped."); |
| } |
| |
| queueInfo = ((ApplicationClient*) appClient)->getQueueInfo(queue, |
| includeApps, includeChildQueues, recursive); |
| return FR_SUCCEEDED; |
| } catch (std::exception& e) { |
| std::stringstream errorMsg; |
| errorMsg << "LibYarnClient::getQueueInfo, Catch the Exception:" << e.what(); |
| setErrorMessage(errorMsg.str()); |
| return FR_FAILED; |
| } catch (...) { |
| std::stringstream errorMsg; |
| errorMsg << "LibYarnClient::getQueueInfo, catch unexpected exception."; |
| setErrorMessage(errorMsg.str()); |
| return FR_FAILED; |
| } |
| } |
| |
| int LibYarnClient::getClusterNodes(list<NodeState> &states,list<NodeReport> &nodeReports) { |
| try{ |
| if (!keepRun && needHeartbeatAlive) { |
| throw std::runtime_error("LibYarnClient::libyarn AM heartbeat thread has stopped."); |
| } |
| nodeReports = ((ApplicationClient*) appClient)->getClusterNodes(states); |
| return FR_SUCCEEDED; |
| } catch (std::exception& e) { |
| std::stringstream errorMsg; |
| errorMsg << "LibYarnClient::getClusterNodes, Catch the Exception:" << e.what(); |
| setErrorMessage(errorMsg.str()); |
| return FR_FAILED; |
| } catch (...) { |
| std::stringstream errorMsg; |
| errorMsg << "LibYarnClient::getClusterNodes, catch unexpected exception."; |
| setErrorMessage(errorMsg.str()); |
| return FR_FAILED; |
| } |
| } |
| } |
