MDFed.cc

Go to the documentation of this file.

/*
  Copyright (c) Members of the EGEE Collaboration. 2004.
  See http://www.eu-egee.org/partners/ for details on the copyright
  holders.

  Licensed 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 "MDFed.h"
#include <iostream>
#include <fstream>
#include "debug.h"

using namespace std;

MDFed::MDFed(bool dbg)
  : debug(dbg), pmdc(NULL)
{
}

MDFed::~MDFed() 
{
}

/************************************************
 * setParentMDClient
 *   Set parent MDClient 
 *   In case of ServerSide Federation, it is NULL. 
 ***********************************************/

void MDFed::setParentMDClient(MDClient* mdc)
{
  pmdc = mdc;
}

/************************************************
 * decodeFedFlags
 *   Retrieve Policies for federation
 *     - cmdType : command type 
 *     - allowFedConnFail : whether to allow connection failure
 *     - allowFedExecFail : whether to allow execution failure
 ***********************************************/

int MDFed::decodeFedFlags(std::string& flags)
{
    char* ptr;
    int flag_int = strtol(flags.c_str(), &ptr, 10);

    decodeFedFlags(flag_int);
}

int MDFed::decodeFedFlags(int flag_int)
{
    cmdType  = flag_int ;
    allowFedConnFail = flag_int & ALLOW_FED_CONN ;
    allowFedExecFail = flag_int & ALLOW_FED_EXEC ;

    DMESG("CMD TYPE : " << cmdType << std::endl);
    DMESG("FED CONN : " << allowFedConnFail << std::endl);
    DMESG("FED EXEC : " << allowFedExecFail << std::endl);
}

/************************************************
 * lookupFedHosts
 *   Check whether there is information about the given host
 *   If it exists, set its status to 1 
 ***********************************************/

int MDFed::lookupFedHosts(std::string &dn) {
    std::list<AMGA::MDChild>::iterator I = mdchild.begin();
    while( I != mdchild.end() ) {
      if( I->dn == dn ) {
        I->status = 1;
        DMESG("Using : " << dn << " : " << I->dn << std::endl);
        return 0;
      } else {
        DMESG("Not Using : " << dn << " : " << I->dn << std::endl);
      }
      I++;
    }
    return -1;
}


/************************************************
 * retrieveFedHosts
 *   Retrieve lists of hosts managing real data
 *   Only called with Client Side Federation
 ***********************************************/

int MDFed::retrieveFedHosts(std::string& errmsg)
{
  int count = 0 ;
  std::string sflags;
  int cmdType;
  int allowConnErr, allowExecErr;

  if( pmdc == NULL ) return -1;
  errmsg = pmdc->getError();

  /* retrieve federation policy */
  if( pmdc->fetchRow(sflags) ) {
    errmsg = "Federation Flag Retrieval Error\n" ;
    return -1;
  }

  decodeFedFlags(sflags);

  while( ! pmdc->eot() ) {
    std::string tmp_dn;
    std::string tmp_host;
    std::string tmp_port;

    /* retrieve host name and port number managing federated data */
    if( pmdc->fetchRow(tmp_host) || pmdc->fetchRow(tmp_port) ) {
      count = 0;
      break;
    }

    tmp_dn = tmp_host;
    tmp_dn.append(":");
    tmp_dn.append(tmp_port);

    /* Check whether the host exists already */
    if ( lookupFedHosts(tmp_dn) > -1 ) {
      count++;
      continue;
    }

    char* ptr;
    int tmp_port_int = strtol(tmp_port.c_str(), &ptr, 10);
    if( tmp_port_int == 0 ) tmp_port_int = MD_DEFAULT_PORT;

    /* Create a new MDClient for new federated hosts */
    MDClient *child = new MDClient(tmp_host, tmp_port_int);
    AMGA::MDChild mdc(tmp_dn, tmp_host, tmp_port_int, (void*) child);

    /* set it to valid */
    mdc.status = 1;
    /* set its connection is not setup yet */
    mdc.connection = 0;
    DMESG("new service2 : " << tmp_dn << std::endl);
    mdchild.push_back(mdc);
    count++;
  }

  if( count == 0 ) {
    errmsg = "Federation Host Retrieval Error\n" ;
    return -1;
  }

  return 0;
}

/************************************************
 * addFedHosts
 *   Insert information about a federated host
 *   Only called with Server Side Federation
 ***********************************************/

int MDFed::addFedHost(std::string tmp_host, int tmp_port, Site & s)
{
  std::string tmp_dn; 
  std::stringstream out_t;

  out_t << tmp_host << ":" << tmp_port;
  tmp_dn = out_t.str();

  /* Check whether the host exists already */
  if ( lookupFedHosts(tmp_dn) > -1 ) 
    return -1;

  char* ptr;

  char fname[L_tmpnam+1];
  tmpnam(fname);
  MDClient* child;

  ofstream myfile;
  myfile.open(fname);
  

  /* Load a site information on the given host 
     , which is mainly related to authentication */
  if (strlen(s.login)) 
    myfile << "Login = " << s.login << "\n";
  if (strlen(s.password)) 
    myfile << "Password = " << s.password << "\n";
  if ( s.useSSL ) 
    myfile << "UseSSL = 1\n";
  else 
    myfile << "UseSSL = 0\n";

  if ( s.authenticateWithCertificate )
    myfile << "AuthenticateWithCertificate = 1\n";
  else
    myfile << "AuthenticateWithCertificate = 0\n";

  if ( strlen(s.certFile) ) 
    myfile << "CertFile = " << s.certFile << "\n";
  if ( strlen(s.keyFile) ) 
    myfile << "KeyFile = " << s.keyFile << "\n";
  if ( s.useGridProxy )
    myfile << "UseGridProxy = 1\n";
  else 
    myfile << "UseGridProxy = 0\n";
  if (s.verifyServerCert)
    myfile << "VerifyServerCert = 1\n";
  else 
    myfile << "VerifyServerCert = 0\n";
  if ( strlen(s.trustedCertDir) ) 
    myfile << "TrustedCertDir = " << s.trustedCertDir << "\n";
  if ( strlen(s.certPassword) )
    myfile << "PrivateKeyPassword = " << s.certPassword << "\n";
  myfile.close();

  /* Create a new MDClient for new federated hosts */
  child = new MDClient(tmp_host, tmp_port, false, fname);
  remove (fname);

  AMGA::MDChild mdc(tmp_dn, tmp_host, tmp_port, (void*) child);

  mdc.status = 1;
  mdc.connection = 0;
  DMESG("new service : " << tmp_dn << std::endl);
  mdchild.push_back(mdc);
  return 0;
}


/************************************************
 * removeFedHeader
 *   In case of SELECT, column information from the second federated server
 *   should be removed. 
 ***********************************************/

int MDFed::removeFedHeader(int col){
  std::string s;

  DMESG("Removing : " << col << " rows" << std::endl);
  for(int i=0; i < col; i++ ) {
    if( fetchRow(s, false) ) {
      return -1;
    }
  }
  return 0;
}

/************************************************
 * initNextFedHosts
 *   Find and connect next federated host, and execute the given command
 *
 *   flags : first call to this function for a command
 ***********************************************/

int MDFed::initNextFedHosts(int flags, std::string& errmsg)
{

    DMESG("initNextFedHosts start ...  " << std::endl);

    std::list<AMGA::MDChild>::iterator I;
    if ( flags ) {
      I = mdchild.begin();
      failureCode = 0;
    } else {
      /* if the type is random, 
           it should have had a successful execution when the flag was true*/
      if ( (cmdType & FED_FILTER) == FED_RANDOM ) return -1;
      I = currentChild ;
      I++;
    }

    int tmp_success = 0;
    while( I != mdchild.end() ) {
      bool justConnected = false;

      // if the host is not a valid one, skip
      if( I->status == 0 ) { I++; continue; }

      DMESG("Service II : " << I->dn << std::endl);
      currentChild = I;

      MDClient *child = (MDClient*) I->mdc;

      // if connection is not made, make a new connection to a host
      if ( I->connection == 0 ) {
        if ( child->connectToServer() ) {
          DMESG("Failed to connect ...  " << cmdType << std::endl);
          errmsg = "failed to access federated directory ";

          // if it allows connection failure, try another server
          if ( allowFedConnFail || ((cmdType & FED_FILTER) == FED_RANDOM) ) {
            DMESG("Failure code 2 " << std::endl);
            failureCode |= 2;
            I++; continue;
          } else {
            errmsg.append("[").append(I->dn).append("]");
            return -1;
          }
        }
        // this variable is set only when connection is just made. 
        justConnected = true;
      }

      // mark the connection is set
      I->connection = 1;

      // execute the command
      if ( child->execute(lastCommand) ) {

        // if connection is made some time agon, let's give another chance
        // if (justConnected == false) { I->connection = 0; continue; }
        errmsg = child->getError();
        DMESG("Execution Failure : " << errmsg << std::endl);

        // get rid of all the remained data at the socket
        if(! child->endOfTransmission) child->fetchData();

        // if it allows execution failure, try another server    
        if ( allowFedExecFail  || ((cmdType & FED_FILTER) == FED_RANDOM) ) {
          DMESG("Failure code 1 " << std::endl);
          failureCode |= 1;
          I++; continue;
        } else {
          errmsg.append("[").append(I->dn).append("]");
          return -1;
        }
      }

      // In case of SELECT, 
      // column information from the second federated server should be removed.
      if( (flags == 0) && (cmdType & FED_SELE))  {
        DMESG("Removing HEADER of Select  " << std::endl);
        if ( removeFedHeader(child->resultCols) < 0 ) {
          errmsg = "Failed to remove Fed Header";
          return -1;
        }
      }

      tmp_success++;
      I++;

      /* FED_ONCE means that the commands should be executed at all servers
         right now */
      if ( (cmdType & FED_ONCE) == 0 )
        return 0;

      /* FED_RANDOM means that it is successful if the command is executed
         successfully */
      if ( (cmdType & FED_FILTER) == FED_RANDOM )
        return 0;
    }
    DMESG("initNextFedHosts end ...  " << std::endl);
    if (tmp_success > 0) return 0;
    return -1;
}

/************************************************
 * clearFedHosts
 *   set the status of all the hosts to be not valid
 ***********************************************/

void MDFed::clearFedHosts() {
    std::list<AMGA::MDChild>::iterator I = mdchild.begin();
    while( I != mdchild.end() ) {
      I->status = 0;
      I++;
    }
}

/************************************************
 * initFed
 *   initilize federation
 *   it is executed whenever a new command is executed 
 *     and its target is federated
 ***********************************************/

int MDFed::initFed(std::string& errmsg, bool serverSide){

    /* In case of client side federation,
       retrieve list of federated hosts from the server */
    if ( (serverSide == 0) && (retrieveFedHosts(errmsg) < 0) ) {
      DMESG("retrieveFedHost Error" << std::endl);
      if ( pmdc != NULL) pmdc->setError(errmsg);
      return 99; 
    }

    // If the command is not supported one by federation
    if( (cmdType & FED_FILTER) == FED_NONE) {
      DMESG("Not allowed Command Error" << std::endl);
      errmsg = "Not allowed command for federation" ;
      if ( pmdc != NULL) pmdc->setError(errmsg);
      return 99; 
    }

    // Find a next federated host, connect and execute the command
    if ( initNextFedHosts(1, errmsg) < 0 ) {
      DMESG("init Next Host Error" << std::endl);
      if ( pmdc != NULL) pmdc->setError(errmsg);
      return 99 ; 
    }

    proxyState = true;
    return 0;
}

/************************************************
 * fetchRow
 *   Fetch a row in terms of federation
 *   If end of data, then go for a next federated host
 ***********************************************/

int MDFed::fetchRow(std::string &line, bool readAhead)
{
    MDClient* child ;
    child = (MDClient*) currentChild->mdc;
    
    int bytes;
    std::string errmsg;
    while ( (bytes = child->fetchRow(line, readAhead)) < 0 ) {
      if ( initNextFedHosts(0,errmsg) < 0 ) {
        return 0;
      }
      child = (MDClient*) currentChild->mdc;
    }
    return bytes;
}


/************************************************
 * eot
 *   Check end of data in terms of federation
 ***********************************************/

int MDFed::eot()
{
    MDClient *child ;
    child = (MDClient*) currentChild->mdc;

    std::string errmsg;
    while (child->eot()) {
      if ( initNextFedHosts(0, errmsg) < 0 ) {
        return true;
      }
      child = (MDClient*) currentChild->mdc;
    }
    return false;
}


/************************************************
 * disconnectFed
 *   disconnect from all federated hosts
 ***********************************************/

void MDFed::disconnectFed(bool saveSession)
{
  std::list<AMGA::MDChild>::iterator I = mdchild.begin();

  MDClient *child ;
  while( I != mdchild.end() ) {
    if( (I->mdc != NULL) && (I->connection == 1) ) {
      child = (MDClient*) I->mdc;
      //child->disconnect(saveSession);
      child->disconnect(false);
      delete child;
    }
    I++;
  }

  mdchild.clear();
}

---