def create_cluster(self, cluster, project_id=None, retry=DEFAULT, timeout=DEFAULT):
        """
        
        """

        if isinstance(cluster, dict):
            cluster_proto = Cluster()
            cluster = self._dict_to_proto(py_dict=cluster, proto=cluster_proto)
        elif not isinstance(cluster, Cluster):
            raise AirflowException(
                "cluster is not instance of Cluster proto or python dict")

        self._append_label(cluster, 'airflow-version', 'v' + version.version)

        self.log.info(
            "Creating (project_id=%s, zone=%s, cluster_name=%s)",
            self.project_id, self.location, cluster.name
        )
        try:
            op = self.get_client().create_cluster(project_id=project_id or self.project_id,
                                                  zone=self.location,
                                                  cluster=cluster,
                                                  retry=retry,
                                                  timeout=timeout)
            op = self.wait_for_operation(op)

            return op.target_link
        except AlreadyExists as error:
            self.log.info('Assuming Success: %s', error.message)
            return self.get_cluster(name=cluster.name).self_link