def set_refresh_interval(self, seconds):
        if seconds <= 0:
            interval = -1
        else:
            interval = unicode(seconds) + "s"

        if self.cluster.version.startswith("0.90."):
            response = self.cluster.put(
                "/" + self.settings.index + "/_settings",
                data='{"index":{"refresh_interval":' + convert.value2json(interval) + '}}'
            )

            result = convert.json2value(utf82unicode(response.all_content))
            if not result.ok:
                Log.error("Can not set refresh interval ({{error}})", {
                    "error": utf82unicode(response.all_content)
                })
        elif any(map(self.cluster.version.startswith, ["1.4.", "1.5.", "1.6.", "1.7."])):
            response = self.cluster.put(
                "/" + self.settings.index + "/_settings",
                data=convert.unicode2utf8('{"index":{"refresh_interval":' + convert.value2json(interval) + '}}')
            )

            result = convert.json2value(utf82unicode(response.all_content))
            if not result.acknowledged:
                Log.error("Can not set refresh interval ({{error}})", {
                    "error": utf82unicode(response.all_content)
                })
        else:
            Log.error("Do not know how to handle ES version {{version}}", version=self.cluster.version)

    def create_index(
        self,
        index,
        alias=None,
        create_timestamp=None,
        schema=None,
        limit_replicas=None,
        read_only=False,
        tjson=False,
        settings=None
    ):
        if not alias:
            alias = settings.alias = settings.index
            index = settings.index = proto_name(alias, create_timestamp)

        if settings.alias == index:
            Log.error("Expecting index name to conform to pattern")

        if settings.schema_file:
            Log.error('schema_file attribute not supported.  Use {"$ref":<filename>} instead')

        if schema == None:
            Log.error("Expecting a schema")
        elif isinstance(schema, basestring):
            schema = convert.json2value(schema, leaves=True)
        else:
            schema = convert.json2value(convert.value2json(schema), leaves=True)

        if limit_replicas:
            # DO NOT ASK FOR TOO MANY REPLICAS
            health = self.get("/_cluster/health")
            if schema.settings.index.number_of_replicas >= health.number_of_nodes:
                Log.warning("Reduced number of replicas: {{from}} requested, {{to}} realized",
                    {"from": schema.settings.index.number_of_replicas},
                    to= health.number_of_nodes - 1
                )
                schema.settings.index.number_of_replicas = health.number_of_nodes - 1

        self.post(
            "/" + index,
            data=schema,
            headers={"Content-Type": "application/json"}
        )

        # CONFIRM INDEX EXISTS
        while True:
            try:
                state = self.get("/_cluster/state", retry={"times": 5}, timeout=3)
                if index in state.metadata.indices:
                    break
                Log.note("Waiting for index {{index}} to appear", index=index)
            except Exception, e:
                Log.warning("Problem while waiting for index {{index}} to appear", index=index, cause=e)
            Thread.sleep(seconds=1)

    def create_index(
        self,
        index,
        alias=None,
        schema=None,
        limit_replicas=None,
        read_only=False,
        tjson=False,
        settings=None
    ):
        if not settings.alias:
            settings.alias = settings.index
            settings.index = proto_name(settings.alias)

        if settings.alias == settings.index:
            Log.error("Expecting index name to conform to pattern")

        if settings.schema_file:
            Log.error('schema_file attribute not supported.  Use {"$ref":<filename>} instead')

        if schema == None:
            Log.error("Expecting a schema")
        elif isinstance(schema, basestring):
            schema = convert.json2value(schema, leaves=True)
        else:
            schema = convert.json2value(convert.value2json(schema), leaves=True)

        if limit_replicas:
            # DO NOT ASK FOR TOO MANY REPLICAS
            health = self.get("/_cluster/health")
            if schema.settings.index.number_of_replicas >= health.number_of_nodes:
                Log.warning("Reduced number of replicas: {{from}} requested, {{to}} realized",
                    {"from": schema.settings.index.number_of_replicas},
                    to= health.number_of_nodes - 1
                )
                schema.settings.index.number_of_replicas = health.number_of_nodes - 1

        self.post(
            "/" + settings.index,
            data=schema,
            headers={"Content-Type": "application/json"}
        )
        while True:
            time.sleep(1)
            try:
                self.head("/" + settings.index)
                break
            except Exception:
                Log.note("{{index}} does not exist yet", index=settings.index)

        es = Index(settings=settings)
        return es

    def test_request(self):
        # SIMPLEST POSSIBLE REQUEST (NOTHING IMPORTANT HAPPENING)
        data = {
            "meta": {
                "suite": "sessionrestore_no_auto_restore osx-10-10",
                "platform": "osx-10-10",
                "e10s": False,
                "och": "opt",
                "bucket": "startup",
                "statistic": "mean"
            },
            "header": ["rownum", "timestamp", "revision", "value"],
            "data": [
                [1, "2015-12-06 09:21:15", "18339318", 879],
                [2, "2015-12-06 16:50:36", "18340858", 976],
                [3, "2015-12-06 19:01:54", "18342319", 880],
                [4, "2015-12-06 21:08:56", "18343567", 1003],
                [5, "2015-12-06 23:33:27", "18345266", 1002],
                [6, "2015-12-07 02:16:22", "18347807", 977],
                [7, "2015-12-07 02:18:29", "18348057", 1035],
                [8, "2015-12-07 04:51:52", "18351263", 1032],
                [9, "2015-12-07 05:29:42", "18351078", 1035],
                [10, "2015-12-07 05:50:37", "18351749", 1010]
            ]
        }

        response = requests.post(settings.url, json=data)
        self.assertEqual(response.status_code, 200)
        data = convert.json2value(convert.utf82unicode(response.content))
        self.assertEqual(data, {})

 def __init__(self, settings):
     self.settings = wrap({"host":"fake", "index":"fake"})
     self.filename = settings.filename
     try:
         self.data = convert.json2value(File(self.filename).read())
     except IOError:
         self.data = Dict()

def process_test_result(source_key, source, destination, please_stop=None):
    path = key2path(source_key)
    destination.delete({"and": [
        {"term": {"etl.source.id": path[1]}},
        {"term": {"etl.source.source.id": path[0]}}
    ]})

    lines = source.read_lines()

    keys = []
    data = []
    for l in lines:
        record = convert.json2value(l)
        if record._id==None:
            continue
        record.result.crash_result = None  #TODO: Remove me after May 2015
        keys.append(record._id)
        data.append({
            "id": record._id,
            "value": record
        })
        record._id = None
    if data:
        try:
            destination.extend(data)
        except Exception, e:
            if "Can not decide on index by build.date" in e:
                if source.bucket.name == "ekyle-test-result":
                    # KNOWN CORRUPTION
                    # TODO: REMOVE LATER (today = Mar2015)
                    delete_list = source.bucket.keys(prefix=key_prefix(source_key))
                    for d in delete_list:
                        source.bucket.delete_key(d)
            Log.error("Can not add to sink", e)

 def __init__(self, filename, host="fake", index="fake", settings=None):
     self.settings = settings
     self.filename = settings.filename
     try:
         self.data = convert.json2value(File(self.filename).read())
     except Exception:
         self.data = Dict()

def get_active_data(settings):
    query = {
    "limit": 100000,
    "from": "unittest",
    "where": {"and": [
        {"eq": {"result.ok": False}},
        {"gt": {"run.timestamp": RECENT.milli}}
    ]},
    "select": [
        "result.ok",
        "build.branch",
        "build.platform",
        "build.release",
        "build.revision",
        "build.type",
        "build.revision",
        "build.date",
        "run.timestamp",
        "run.suite",
        "run.chunk",
        "result.test",
        "run.stats.status.test_status"
    ],
    "format": "table"
    }
    result = http.post("http://activedata.allizom.org/query", data=convert.unicode2utf8(convert.value2json(query)))

    query_result = convert.json2value(convert.utf82unicode(result.all_content))

    tab = convert.table2tab(query_result.header, query_result.data)
    File(settings.output.activedata).write(tab)

def process(source_key, source, destination, resources, please_stop=None):
    lines = source.read_lines()

    etl_header = convert.json2value(lines[0])
    if etl_header.etl:
        start = 0
    elif etl_header.locale or etl_header._meta:
        start = 0
    else:
        start = 1

    keys = []
    records = []
    stats = Dict()
    for i, line in enumerate(lines[start:]):
        pulse_record = Null
        try:
            pulse_record = scrub_pulse_record(source_key, i, line, stats)
            if not pulse_record:
                continue

            with Profiler("transform_buildbot"):
                record = transform_buildbot(pulse_record.payload, resources=resources)
                record.etl = {
                    "id": i,
                    "source": pulse_record.etl,
                    "type": "join",
                    "revision": get_git_revision()
                }
            key = etl2key(record.etl)
            keys.append(key)
            records.append({"id": key, "value": record})
        except Exception, e:
            Log.warning("Problem with pulse payload {{pulse|json}}", pulse=pulse_record.payload, cause=e)

def solve():
    try:
        data = convert.json2value(convert.utf82unicode(flask.request.data))
        solved = noop.solve(data)
        response_data = convert.unicode2utf8(convert.value2json(solved))

        return Response(
            response_data,
            direct_passthrough=True,  # FOR STREAMING
            status=200,
            headers={
                "access-control-allow-origin": "*",
                "content-type": "application/json"
            }
        )
    except Exception, e:
        e = Except.wrap(e)
        Log.warning("Could not process", cause=e)
        e = e.as_dict()
        return Response(
            convert.unicode2utf8(convert.value2json(e)),
            status=400,
            headers={
                "access-control-allow-origin": "*",
                "content-type": "application/json"
            }
        )

def get_env(ref, url):
    # GET ENVIRONMENT VARIABLES
    ref = ref.host
    try:
        new_value = _convert.json2value(os.environ[ref])
    except Exception, e:
        new_value = os.environ[ref]

def get_json(url, **kwargs):
    """
    ASSUME RESPONSE IN IN JSON
    """
    response = get(url, **kwargs)
    c = response.all_content
    return convert.json2value(convert.utf82unicode(c))

    def pop(self, wait=SECOND, till=None):
        m = self.queue.read(wait_time_seconds=Math.floor(wait.seconds))
        if not m:
            return None

        self.pending.append(m)
        return convert.json2value(m.get_body())

    def test_request(self):
        # MAKE SOME DATA
        data = {
            "constant": "this is a test",
            "random-data": convert.bytes2base64(Random.bytes(100))
        }

        client = Client(settings.url, unwrap(settings.hawk))  # unwrap() DUE TO BUG https://github.com/kumar303/mohawk/issues/21
        link, id = client.send(data)
        Log.note("Success!  Located at {{link}} id={{id}}", link=link, id=id)

        # FILL THE REST OF THE FILE
        Log.note("Add ing {{num}} more...", num=99-id)
        for i in range(id + 1, storage.BATCH_SIZE):
            l, k = client.send(data)
            if l != link:
                Log.error("Expecting rest of data to have same link")

        # TEST LINK HAS DATA
        raw_content = requests.get(link).content
        content = convert.zip2bytes(raw_content)
        for line in convert.utf82unicode(content).split("\n"):
            data = convert.json2value(line)
            if data.etl.id == id:
                Log.note("Data {{id}} found", id=id)
                break
        else:
            Log.error("Expecting to find data at link")

    def delete_index(self, index_name):
        if not isinstance(index_name, unicode):
            Log.error("expecting an index name")

        if self.debug:
            Log.note("Deleting index {{index}}", index=index_name)

        # REMOVE ALL ALIASES TOO
        aliases = [a for a in self.get_aliases() if a.index == index_name and a.alias != None]
        if aliases:
            self.post(
                path="/_aliases",
                data={"actions": [{"remove": a} for a in aliases]}
            )

        url = self.settings.host + ":" + unicode(self.settings.port) + "/" + index_name
        try:
            response = http.delete(url)
            if response.status_code != 200:
                Log.error("Expecting a 200")
            details = convert.json2value(utf82unicode(response.content))
            if self.debug:
                Log.note("delete response {{response}}", response=details)
            return response
        except Exception, e:
            Log.error("Problem with call to {{url}}", url=url, cause=e)

def _get_url(url, branch, **kwargs):
    with Explanation("get push from {{url}}", url=url):
        response = http.get(url, **kwargs)
        data = convert.json2value(response.content.decode("utf8"))
        if isinstance(data, basestring) and data.startswith("unknown revision"):
            Log.error("Unknown push {{revision}}", revision=strings.between(data, "'", "'"))
        branch.url = _trim(url)  #RECORD THIS SUCCESS IN THE BRANCH
        return data

def expand_json(rows):
    # CONVERT JSON TO VALUES
    for r in rows:
        for k, json in list(r.items()):
            if isinstance(json, basestring) and json[0:1] in ("[", "{"):
                with suppress_exception:
                    value = convert.json2value(json)
                    r[k] = value

def post_json(url, **kwargs):
    """
    ASSUME RESPONSE IN IN JSON
    """
    kwargs["data"] = convert.unicode2utf8(convert.value2json(kwargs["data"]))

    response = post(url, **kwargs)
    c=response.all_content
    return convert.json2value(convert.utf82unicode(c))

 def delete(self, path, **kwargs):
     url = self.settings.host + ":" + unicode(self.settings.port) + path
     try:
         response = convert.json2value(utf82unicode(http.delete(url, **kwargs).content))
         if self.debug:
             Log.note("delete response {{response}}",  response= response)
         return response
     except Exception, e:
         Log.error("Problem with call to {{url}}",  url= url, cause=e)

    def read_json(self, encoding="utf8"):
        from pyLibrary.jsons import ref

        content = self.read(encoding=encoding)
        value = convert.json2value(content, flexible=True, leaves=True)
        abspath = self.abspath
        if os.sep == "\\":
            abspath = "/" + abspath.replace(os.sep, "/")
        return ref.expand(value, "file://" + abspath)