func (zom *zookeeperOffsetManager) commitOffset(topic string, partition int32, tracker *partitionOffsetTracker, logger zap.Logger) error {
	err := tracker.commit(func(offset int64) error {
		if offset >= 0 {
			return zom.cg.group.CommitOffset(topic, partition, offset+1)
		} else {
			return nil
		}
	})

	if err != nil {
		logger.Warn("ZOOKEEPER: FAILED to commit offset",
			zap.Int64("highestProcessedOffset", tracker.highestProcessedOffset),
			zap.String("topic", topic),
			zap.Int64("partition", int64(partition)),
		)
	} else if zom.config.VerboseLogging {
		logger.Debug("ZOOKEEPER: Committed offset",
			zap.Int64("lastCommittedOffset", tracker.lastCommittedOffset),
			zap.String("topic", topic),
			zap.Int64("partition", int64(partition)),
		)
	}

	return err
}

func main() {
	logLevel := zap.LevelFlag("v", zap.InfoLevel, "log level: all, debug, info, warn, error, panic, fatal, none")
	flag.StringVar(&botName, "botname", "satpam_bot", "bot name")
	flag.StringVar(&adminID, "admin", "", "admin id")
	flag.Parse()

	// setup logger
	log.SetLevel(*logLevel)
	bot.SetLogger(log)
	log.Info("STARTED", zap.String("version", VERSION), zap.String("buildtime", BUILDTIME))

	key := os.Getenv("TELEGRAM_KEY")
	if key == "" {
		log.Fatal("TELEGRAM_KEY can not be empty")
	}

	startedAt = time.Now()
	telegram := bot.NewTelegram(key)
	plugin := satpamBot{t: telegram}
	if err := telegram.AddPlugin(&plugin); err != nil {
		log.Fatal("Failed AddPlugin", zap.Error(err))
	}
	plugin.start()
	telegram.Start()

}

// WrongURLsToLog - write to log add wrong URLs
func (h *HTMLMetadata) WrongURLsToLog(logger zap.Logger) {
	for url, error := range h.wrongURLs {
		logger.Warn("Error parse URL",
			zap.String("err_url", url),
			zap.String("details", error),
		)
	}
}

func (k KeeperInfoHistories) DeepCopy() KeeperInfoHistories {
	if k == nil {
		return nil
	}
	nk, err := copystructure.Copy(k)
	if err != nil {
		panic(err)
	}
	log.Debug("", zap.String("k", spew.Sdump(k)), zap.String("nk", spew.Sdump(nk)))
	if !reflect.DeepEqual(k, nk) {
		panic("not equal")
	}
	return nk.(KeeperInfoHistories)
}

func (s *Server) serveH2c(w http.ResponseWriter, r *http.Request) {
	defer func() {
		if err := recover(); err != nil {
			w.WriteHeader(http.StatusInternalServerError)
			Log.Error("CONNECT failed", zap.Object("err", err))
		}
	}()
	remote, err := net.DialTimeout("tcp", r.Host, time.Second*10)
	if err != nil {
		Log.Error("dail failed", zap.Error(err), zap.String("host", r.Host))
		w.WriteHeader(http.StatusNotImplemented)
		return
	}
	defer remote.Close()

	fw := &flushWriter{w}
	fw.FlushHeader(http.StatusOK)
	go io.Copy(remote, r.Body)
	srcRemote := &TryReader{
		c:        remote,
		ignore:   3,
		maxRetry: 2,
		tryDur:   time.Millisecond * 600,
		timeout:  time.Second * 15,
	}
	io.Copy(fw, srcRemote)
}

func sentinel(cmd *cobra.Command, args []string) {
	if cfg.debug {
		log.SetLevel(zap.DebugLevel)
	}
	if cfg.clusterName == "" {
		fmt.Println("cluster name required")
		os.Exit(1)
	}
	if cfg.storeBackend == "" {
		fmt.Println("store backend type required")
		os.Exit(1)

	}

	id := common.UID()
	log.Info("sentinel id", zap.String("id", id))

	stop := make(chan bool, 0)
	end := make(chan bool, 0)
	sigs := make(chan os.Signal, 1)
	signal.Notify(sigs, os.Interrupt, os.Kill)
	go sigHandler(sigs, stop)

	s, err := NewSentinel(id, &cfg, stop, end)
	if err != nil {
		fmt.Printf("cannot create sentinel: %v\n", err)
		os.Exit(1)
	}
	go s.Start()

	<-end
}

func (p *Manager) SyncFromFollowedPGRewind(followedConnParams ConnParams, password string) error {
	// ioutil.Tempfile already creates files with 0600 permissions
	pgpass, err := ioutil.TempFile("", "pgpass")
	if err != nil {
		return err
	}
	defer os.Remove(pgpass.Name())
	defer pgpass.Close()

	host := followedConnParams.Get("host")
	port := followedConnParams.Get("port")
	user := followedConnParams.Get("user")
	pgpass.WriteString(fmt.Sprintf("%s:%s:*:%s:%s\n", host, port, user, password))

	// Disable synchronous commits. pg_rewind needs to create a
	// temporary table on the master but if synchronous replication is
	// enabled and there're no active standbys it will hang.
	followedConnParams.Set("options", "-c synchronous_commit=off")
	followedConnString := followedConnParams.ConnString()

	log.Info("running pg_rewind")
	name := filepath.Join(p.pgBinPath, "pg_rewind")
	cmd := exec.Command(name, "--debug", "-D", p.dataDir, "--source-server="+followedConnString)
	cmd.Env = append(cmd.Env, fmt.Sprintf("PGPASSFILE=%s", pgpass.Name()))
	log.Debug("execing cmd", zap.Object("cmd", cmd))
	out, err := cmd.CombinedOutput()
	if err != nil {
		return fmt.Errorf("error: %v, output: %s", err, string(out))
	}
	log.Debug("cmd out", zap.String("out", string(out)))
	return nil
}

func (s *Server) serveH2r(w http.ResponseWriter, r *http.Request) {
	defer func() {
		if err := recover(); err != nil {
			w.WriteHeader(http.StatusInternalServerError)
			Log.Error("REVERSE failed", zap.Object("err", err))
		} else {
			w.WriteHeader(http.StatusOK)
		}
	}()

	remote, err := net.DialTimeout("tcp", r.Host, time.Second*10)
	if err != nil {
		Log.Error("dail failed", zap.Error(err), zap.String("host", r.Host))
		w.WriteHeader(http.StatusNotImplemented)
		return
	}
	defer remote.Close()

	go io.Copy(remote, r.Body)
	//	go io.Copy(remote, io.TeeReader(r.Body, os.Stdout))
	resr := io.TeeReader(remote, w)
	//	resr = io.TeeReader(resr, os.Stdout)
	res, err := http.ReadResponse(bufio.NewReader(resr), nil)
	if err != nil {
		return
	}
	if res.Body != nil {
		defer res.Body.Close()
		io.Copy(ioutil.Discard, res.Body)
	}
}

func (w *hostWorkers) Init(db *content.DBrw, logger zap.Logger, baseHosts []string, cnt int) error {
	hostMng := &hostsManager{}
	err := hostMng.Init(db, baseHosts)
	if err != nil {
		return err
	}

	hosts := hostMng.GetHosts()
	w.workers = make([]*hostWorker, 0)
	cntPerHost := cnt / len(hosts)
	if cntPerHost < 1 {
		cntPerHost = 1
	}
	for hostName, hostID := range hosts {
		worker := &hostWorker{Request: &request{hostMng: hostMng}}
		worker.Request.Init(logger.With(zap.String("host", hostName)))
		worker.Tasks, err = db.GetNewURLs(hostID, cntPerHost)
		if err != nil {
			return err
		}
		w.workers = append(w.workers, worker)
	}

	return nil
}

// TestLogError ...
func TestLogError(t *testing.T) {
	Convey("Check Log standart error", t, func() {
		buf := &bytes.Buffer{}
		logger := zap.NewJSON(zap.DebugLevel, zap.Output(zap.AddSync(buf)))
		logger.StubTime()

		LogError(logger, errors.New("message"))
		So(string(buf.Bytes()), ShouldEqual, `{"msg":"message","level":"error","ts":0,"fields":{}}`+"\n")
	})

	Convey("Check Log ErrorEx", t, func() {
		callerTest = true
		defer func() { callerTest = false }()

		buf := &bytes.Buffer{}
		logger := zap.NewJSON(zap.DebugLevel, zap.Output(zap.AddSync(buf)))
		logger.StubTime()

		LogError(logger, NewEx(zap.DebugLevel, "message",
			zap.String("field1", "field1 data"),
			zap.Int("field2", 5)))

		So(string(buf.Bytes()), ShouldEqual,
			`{"msg":"message","level":"debug","ts":0,"fields":{"caller":"<fake>","field1":"field1 data","field2":5}}`+"\n")
	})

}

func readBody(contentEncoding string, body io.Reader) ([]byte, error) {
	var err error
	result := []byte{}
	if contentEncoding == "gzip" {
		reader, err := gzip.NewReader(body)
		if err != nil {
			return result, werrors.NewDetails(ErrReadGZipResponse, err)
		}
		result, err = ioutil.ReadAll(reader)
		if err == nil {
			err = reader.Close()
		} else {
			_ = reader.Close()
		}
		if err != nil {
			return result, werrors.NewDetails(ErrReadGZipResponse, err)
		}
	} else if contentEncoding == "identity" || contentEncoding == "" {
		result, err = ioutil.ReadAll(body)
		if err != nil {
			return result, werrors.NewDetails(ErrReadResponse, err)
		}
	} else {
		return result, werrors.NewFields(ErrUnknownContentEncoding, zap.String("encoding", contentEncoding))
	}

	return result, nil
}

func proxy(cmd *cobra.Command, args []string) {
	if cfg.debug {
		log.SetLevel(zap.DebugLevel)
	}
	stdlog, _ := zwrap.Standardize(log, zap.DebugLevel)
	pollon.SetLogger(stdlog)

	if cfg.clusterName == "" {
		fmt.Printf("cluster name required")
		os.Exit(1)
	}
	if cfg.storeBackend == "" {
		fmt.Printf("store backend type required")
		os.Exit(1)
	}

	uid := common.UID()
	log.Info("proxy uid", zap.String("uid", uid))

	clusterChecker, err := NewClusterChecker(uid, cfg)
	if err != nil {
		fmt.Printf("cannot create cluster checker: %v", err)
		os.Exit(1)
	}
	clusterChecker.Start()
}

func (p *Manager) SetupRoles() error {
	ctx, cancel := context.WithTimeout(context.Background(), p.requestTimeout)
	defer cancel()

	if p.suUsername == p.replUsername {
		log.Info("adding replication role to superuser")
		if err := alterRole(ctx, p.localConnParams, []string{"replication"}, p.suUsername, p.suPassword); err != nil {
			return fmt.Errorf("error adding replication role to superuser: %v", err)
		}
		log.Info("replication role added to superuser")
	} else {
		// Configure superuser role password
		if p.suPassword != "" {
			log.Info("setting superuser password")
			if err := setPassword(ctx, p.localConnParams, p.suUsername, p.suPassword); err != nil {
				return fmt.Errorf("error setting superuser password: %v", err)
			}
			log.Info("superuser password set")
		}
		roles := []string{"login", "replication"}
		log.Info("creating replication role")
		if err := createRole(ctx, p.localConnParams, roles, p.replUsername, p.replPassword); err != nil {
			return fmt.Errorf("error creating replication role: %v", err)
		}
		log.Info("replication role created", zap.String("role", p.replUsername))
	}
	return nil
}

func (p *PostgresKeeper) getLastPGState() *cluster.PostgresState {
	p.pgStateMutex.Lock()
	pgState := p.lastPGState.DeepCopy()
	p.pgStateMutex.Unlock()
	log.Debug("pgstate dump", zap.String("pgState", spew.Sdump(pgState)))
	return pgState
}

func (zom *zookeeperOffsetManager) FinalizePartition(topic string, partition int32, lastOffset int64, timeout time.Duration, replicaId int, logger zap.Logger) error {
	zom.l.RLock()
	tracker := zom.offsets[topic][partition]
	zom.l.RUnlock()

	if lastOffset >= 0 {
		if lastOffset-tracker.highestProcessedOffset > 0 {
			logger.Info("ZOOKEEPER: Finalizing partition. Waiting before processing remaining messages",
				zap.Int("replicaId", replicaId),
				zap.String("topic", topic),
				zap.Int64("partition", int64(partition)),
				zap.Int64("lastProcessedOffset", tracker.highestProcessedOffset),
				zap.Duration("waitingTimeToProcessMoreMessages", timeout/time.Second),
				zap.Int64("numMessagesToProcess", lastOffset-tracker.highestProcessedOffset),
			)
			if !tracker.waitForOffset(lastOffset, timeout) {
				return fmt.Errorf("REP %d - TIMEOUT waiting for offset %d. Last committed offset: %d", replicaId, lastOffset, tracker.lastCommittedOffset)
			}
		}

		if err := zom.commitOffset(topic, partition, tracker, logger); err != nil {
			return fmt.Errorf("REP %d - FAILED to commit offset %d to Zookeeper. Last committed offset: %d", replicaId, tracker.highestProcessedOffset, tracker.lastCommittedOffset)
		}
	}

	zom.l.Lock()
	delete(zom.offsets[topic], partition)
	zom.l.Unlock()

	return nil
}

func main() {
	mongoSession, err := mgo.Dial("127.0.0.1")
	if err != nil {
		panic(err)
	}
	defer mongoSession.Close()

	mongoSession.SetMode(mgo.Monotonic, true)

	log := zap.NewJSON(
		zap.Debug,
		zap.Fields(zap.Int("count", 1)),
		zap.Output(NewWriter(mongoSession)),
	)
	url := "http://example.local"
	tryNum := 42
	startTime := time.Now()
	for i := range [logCounts]struct{}{} {
		log.Info("Failed to fetch URL.",
			zap.String("url", url),
			zap.Int("attempt", tryNum),
			zap.Duration("backoff", time.Since(startTime)),
			zap.Int("index", i),
		)
	}
	fmt.Printf("Finished in %v\n", time.Since(startTime))
}

// NewHTMLMetadata - create new HTMLMetadata struct
func NewHTMLMetadata(hostMng *hostsManager, urlStr string) (*HTMLMetadata, error) {
	baseURL, err := url.Parse(urlStr)
	if err != nil {
		return nil, werrors.NewFields(ErrParseBaseURL,
			zap.String("details", err.Error()),
			zap.String("parsed_url", urlStr))
	}

	return &HTMLMetadata{
		URLs:         make(map[string]sql.NullInt64),
		wrongURLs:    make(map[string]string),
		title:        "",
		MetaTagIndex: true,
		baseURL:      baseURL,
		hostMng:      hostMng,
	}, nil
}

func (m *hostsManager) resolveURL(hostName string) (string, error) {
	hostURL := NormalizeURL(&url.URL{Scheme: "http", Host: hostName})
	response, err := http.Get(hostURL)
	if err == nil {
		err = response.Body.Close()
		if response.StatusCode != 200 {
			return "", werrors.NewFields(ErrResolveBaseURL,
				zap.Int("status_code", response.StatusCode),
				zap.String("url", hostURL))
		}
	}
	if err != nil {
		return "", werrors.NewFields(ErrGetRequest,
			zap.String("details", err.Error()),
			zap.String("url", hostURL))
	}

	return response.Request.URL.String(), nil
}

func (m *hostsManager) initByDb(db proxy.DbHost) error {
	hosts, err := db.GetHosts()
	if err != nil {
		return err
	}
	for id, host := range hosts {
		hostName := host.GetName()
		robot, err := robotstxt.FromStatusAndBytes(host.GetRobotsTxt())
		if err != nil {
			return werrors.NewFields(ErrCreateRobotsTxtFromDb,
				zap.String("host", hostName),
				zap.String("details", err.Error()))
		}
		m.hosts[hostName] = id
		m.robotsTxt[id] = robot.FindGroup("Googlebot")
	}

	return nil
}

func (m *hostsManager) readRobotTxt(hostName string) (int, []byte, error) {
	var body []byte
	robotsURL := NormalizeURL(&url.URL{Scheme: "http", Host: hostName, Path: "robots.txt"})
	response, err := http.Get(robotsURL)
	if err == nil {
		body, err = ioutil.ReadAll(response.Body)
		closeErr := response.Body.Close()
		if err == nil {
			err = closeErr
		}
	}

	if err != nil {
		return 0, body, werrors.NewFields(ErrGetRequest,
			zap.String("details", err.Error()),
			zap.String("url", robotsURL))
	}

	return response.StatusCode, body, nil
}