manualmove

manualmove.py

import os
import sys
import getopt
import exceptions
import struct
from optparse import OptionParser
from couchbase.rest_client import RestConnection

parser = OptionParser()

parser.add_option("-c", "--cluster", dest="cluster", default="127.0.0.1")
parser.add_option("-u", "--username", dest="username", default="Administrator")
parser.add_option("-p", "--password", dest="password", default="password")
parser.add_option("-s", "--source",dest="source",default="127.0.0.1:11210")
parser.add_option("-d", "--destination",dest="destination",default="127.0.0.1:11210")
parser.add_option("-b", "--bucket-name", dest="bucket", default="default")
parser.add_option("--sasl", dest="sasl", default="password")
parser.add_option("-v","--vbuckets", dest="vbuckets")
parser.add_option("--dont-stop-orchestrator", dest="dont_stop_orchestrator", action="store_true")
parser.add_option("--dont-restart-orchestrator", dest="dont_restart_orchestrator", action="store_true")
parser.add_option("--only-update-vbucket-map", dest="only_update_vbucket_map", action="store_true")

options, args = parser.parse_args()
cluster_info = {"ip":options.cluster, "port":8091, "username":options.username, "password":options.password}
rest = RestConnection(cluster_info)
nodes = rest.node_statuses()

sourceOtp = ""
sourceMemcached = 11210
destinationOtp = ""
destinationMemcached = 11210
sourceVbuckets = []
print options.bucket
print options.vbuckets.split(",")
for node in nodes:
   if node.ip == options.source:
      sourceOtp = node.id
   if node.ip == options.destination:
      destinationOtp = node.id

print destinationOtp,destinationMemcached
print sourceOtp,sourceMemcached

source_vbuckets = []
vbuckets = rest.get_vbuckets(options.bucket)
selected_vbuckets = []
if options.vbuckets:
   selected_vbuckets = options.vbuckets.split(",")
for vbucket in vbuckets:
   if str(vbucket.id) in selected_vbuckets:
      print vbucket.id,vbucket.master,options.source
      if vbucket.master.find(options.source) >= 0:
         source_vbuckets.append(vbucket.id) 
for s in source_vbuckets:
   print s
#shut down orchestrator
print options.dont_stop_orchestrator
print options.dont_restart_orchestrator
if not options.dont_stop_orchestrator:
   print "stop orchestrator"
   code = "M = mb_master:master_node(), ok = supervisor:terminate_child({mb_master_sup, M}, ns_orchestrator), ok = supervisor:delete_child({mb_master_sup, M}, ns_orchestrator)."
   print code
   api = '{0}{1}'.format(rest.baseUrl, 'diag/eval/')
   status, content = rest._http_request(api, "POST", code)
   print "/diag/eval : status : {0} content : {1}".format(status, content)

#move vbucket
for vbucket in source_vbuckets:
   print "going to move vbucket ",vbucket
   command = ""
#/opt/couchbase/bin/ebucketmigrator -h 10.1.3.67:11210 -A -V -t -b 0 -d 10.1.3.71:11210
   if options.bucket != "default":
      command = "echo {0}| PATH=/opt/couchbase/bin:$PATH /opt/couchbase/bin/ebucketmigrator -h {1}:{2} -A -V -t -b {3} -d {4}:{5} -a {6} -v".format(options.sasl,options.source,sourceMemcached,vbucket,options.destination,destinationMemcached,options.bucket)
   else:
      command = "PATH=/opt/couchbase/bin:$PATH /opt/couchbase/bin/ebucketmigrator -h {0}:{1} -A -V -t -b {2} -d {3}:{4}".format(options.source,sourceMemcached,vbucket,options.destination,destinationMemcached)

#   command = "PATH=/opt/couchbase/bin:$PATH /opt/couchbase/bin/ebucketmigrator -h {0}:{1} -A -V -t -b {2} -d {3}:{4}".format(options.source,sourceMemcached,vbucket,options.destination,destinationMemcached)
   print command
   output = os.popen(command).read()
   print str(output)
   print output.find("successfull")
   if str(output).find("successfull") >= 0:

#update vbucket map
      update_map_code = "{ok, C} = ns_bucket:get_bucket(\"%s\"), V = %s, M = proplists:get_value(map, C), Chain = lists:nth(V+1, M), NewChain = ['%s' | tl(Chain)], NewMap = erlang:tuple_to_list(erlang:setelement(V+1, erlang:list_to_tuple(M), NewChain)), ns_bucket:update_bucket_props(\"%s\", [{map, NewMap}])." % (options.bucket,vbucket,destinationOtp,options.bucket)
      print update_map_code

      api = '{0}{1}'.format(rest.baseUrl, 'diag/eval/')
      status, content = rest._http_request(api, "POST", update_map_code)
      print "/diag/eval : status : {0} content : {1}".format(status, content)
   else:
      print "run vbuketctl to delete the vbucket:{0} from {1}".format(vbucket,options.destination)

print options.only_update_vbucket_map
if options.only_update_vbucket_map:
   for vbucket in selected_vbuckets:
      update_map_code = "{ok, C} = ns_bucket:get_bucket(\"%s\"), V = %s, M = proplists:get_value(map, C), Chain = lists:nth(V+1, M), NewChain = ['%s' | tl(Chain)], NewMap = erlang:tuple_to_list(erlang:setelement(V+1, erlang:list_to_tuple(M), NewChain)), ns_bucket:update_bucket_props(\"%s\", [{map, NewMap}])." % (options.bucket,vbucket,destinationOtp,options.bucket)

      print update_map_code

      api = '{0}{1}'.format(rest.baseUrl, 'diag/eval/')
      status, content = rest._http_request(api, "POST", update_map_code)
      print "/diag/eval : status : {0} content : {1}".format(status, content)

#restart orchestrator
if not options.dont_restart_orchestrator: 
   print "start orchestrator"
   restart_code = "M = mb_master:master_node(), rpc:call(M, erlang,apply, [fun () -> exit(whereis(mb_master), kill) end, []])."
   print restart_code
   api = '{0}{1}'.format(rest.baseUrl, 'diag/eval/')
   status, content = rest._http_request(api, "POST", restart_code)
   print "/diag/eval : status : {0} content : {1}".format(status, content)