Кастомный сервер конвертации файлов для Битрикс24

Пару слов обо мне

Меня зовут Дмитрий, я являюсь PHP разработчиком. Работаю с Битрикс24, Laravel и Go.

Проблема Битрикса

Как известно, рекомендуемое окружение для Битрикса – их собственная разработка BitrixVM на базе CentOS. Иногда такое окружение не устраивает заказчиков, поэтому выбирают Docker или сервер с установленным LEMP стеком. 

При переходе на окружение отличное от BitrixVM, существует две основные проблемы – отсутствие сервера очередей Push&Pull и сервера конвертации файлов. 

Первая проблема решаема: на просторах гитхаба уже существует рабочее решение для развертывания локального сервера в Docker. А также можно использовать облачный сервер, так как для его работы не требуется, чтобы портал был доступен извне. 

Со второй проблемой облако уже не всегда возможно использовать по ряду причин:

• необходимость доступности сайта извне (не подходит для полностью закрытых окружений);

• опасения заказчиков по поводу передачи конфиденциальных документов на облачный сервер.

Единственное решение – установка BitrixVM на отдельном сервере/в докере, с развертыванием бэкапа внутри и использование штатного сервера, встроенного в окружение, что далеко не всегда удобно.

С этим сталкивался и я на закрытых окружениях крупных российских заказчиков, которые далеко не всегда соглашались открывать доступ к порталу из-за соображений безопасности.

После нескольких таких кейсов было решено написать свой полноценный аналог, который бы отвечал нескольким критериям:

• Работал в докере и запускался в несколько кликов;

• Была возможность легко задать количество воркеров под различную нагрузку.

В качестве языка программирования было решено выбрать Go.

Реализация

Для начала нам необходимо понять, что представляет собой штатный сервер конвертации: по сути, это не более чем:

• API сервер с одним эндпоинтом;

• RabbitMQ сервер;

• Consumer для обработки заданий из очереди, со следующим установленным ПО: LibreOffice, MPEG, ImageMagic.

Посмотреть реализацию на PHP можно в исходниках модуля Битрикса. Для упрощения я покажу только основной код, остальное всегда можно будет посмотреть в репозитории.

API сервер

Для API мы будем использовать библиотеку go-chi/chi.

Реализуем единственный обработчик – convert, с преобразованием данных в структуру ConvertTask и последующим добавлением в одну из очередей:

• main_preview – используется в предпросмотре файлов;

• documentgenerator_create – используется в генераторе документов CRM.

// internal/http-server/handlers/convert/convert.go
func New(ctx context.Context, log *slog.Logger, rabbit *rabbitmq.Rabbit) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {

		const op = "handlers.convert.New"

		reqId := middleware.GetReqID(r.Context())

		log = log.With(
			slog.String("op", op),
			slog.String("request_id", reqId),
		)

		err := r.ParseForm()

		if err != nil {
			log.Error("failed to decode request body", sl.Err(err))
			render.JSON(w, r, resp.Error("failed to decode request body", 152))
		}

		task, err := prepareOptions(r.Form, reqId)

		if err != nil {
			log.Error("failed to prepare options", sl.Err(err))
			render.JSON(w, r, resp.Error("failed to parse task", 0))
			return
		}

		if task.Queue == "" {
			task.Queue = rabbit.DefaultQueue()
			log.Warn("not found queue. Set default", slog.String("default_queue", task.Queue))
		}

		taskMsg, err := json.Marshal(task)

		if err != nil {
			log.Error("Error parse request", sl.Err(err))

			render.JSON(w, r, resp.Error("Error parse request", 0))

			return
		}

		err = rabbit.Publish(task.Queue, taskMsg)
		if err != nil {
			log.Error("error publish task", slog.String("queue", task.Queue), sl.Err(err))
			render.JSON(w, r, resp.Error("error publish task", 0))
		}
		render.JSON(w, r, resp.Success())
	}
}

FileUploader 

Создадим структуру FileUploader и реализуем следующие базовые методы:

• Download – скачивание конвертируемого файла из Б24;

• uploadFile – загрузка готового файла в Б24;

• Complete – отправка запроса о завершении конвертации;

• getUploadInfo – получение информации о загружаемом файле.

  Для каждого запроса заводим структуры. Оборачиваем запросы в функцию retry.Do из библиотеки avast/retry-go. Устанавливаем 3 попытки на подключение (на случай сетевых проблем).

// internal/lib/fileuploader/fileuploader.go
Пример метода получения информации о файле
func (f *FileUploader) getUploadInfo(file string, key string) (*uploadInfoResp, error) {
	fileInfo, err := os.Stat(file)

	if err != nil {
		return nil, fmt.Errorf("error get file info [%s]: [%w]", file, err)
	}

	uploadReq := uploadInfoRequest{
		FileId:   key,
		FileSize: fileInfo.Size(),
		Upload:   "where",
	}

	v, err := query.Values(uploadReq)

	if err != nil {
		return nil, fmt.Errorf("error convert struct request to query: [%w]", err)
	}

	res, err := http.PostForm(f.url, v)

	if err != nil {
		return nil, fmt.Errorf("error get upload info from [%s]: [%w]", f.url, err)
	}

	var uploadInfoRes uploadInfoResp

	body, err := io.ReadAll(res.Body)

	if err != nil {
		return nil, fmt.Errorf("wrong response upload info request to url [%s]: [%w]", f.url, err)
	}

	if err = json.Unmarshal(body, &uploadInfoRes); err != nil {
		return nil, fmt.Errorf("error unmarshal upload info request to url [%s]: [%w]", f.url, err)
	}

	return &uploadInfoRes, nil
}

Клиент RabbitMQ

Создаем структуру RabbitMQ и реализуем базовые методы:

• Connect - подключение к rabbitMQ;

• Reconnect - реконнект в случае разрыва соединения, запускаем в отдельной горутине;

• InitQueue - инициализация основной очереди и Dead Letter;

• Consume - чтение сообщений из очереди;

• Publish - отправка сообщения в очередь.

// internal/lib/rabbitmq/rabbitmq.go
package rabbitmq

import (
	"bitrix-converter/internal/config"
	"bitrix-converter/internal/lib/logger/sl"
	"context"
	"fmt"
	amqp "github.com/rabbitmq/amqp091-go"
	"log/slog"
	"time"
)

type Rabbit struct {
	conn *amqp.Connection
	log  *slog.Logger
	cfg  config.RabbitConfig
}

func New(log *slog.Logger, cfg config.RabbitConfig) *Rabbit {
	return &Rabbit{
		log: log,
		cfg: cfg,
	}
}

func (r *Rabbit) DefaultQueue() string {
	return r.cfg.DefaultQueue
}

func (r *Rabbit) Connect() error {
	url := fmt.Sprintf("amqp://%s:%s@%s:%s", r.cfg.User, r.cfg.Password, r.cfg.Host, r.cfg.Port)

	conn, err := amqp.Dial(url)
	if err != nil {
		return fmt.Errorf("failed to connect to RabbitMQ: [%w]", err)
	}

	r.conn = conn
	return nil
}

func (r *Rabbit) Channel() (*amqp.Channel, error) {
	return r.conn.Channel()
}

func (r *Rabbit) Reconnect() {
	for {
		_, ok := <-r.conn.NotifyClose(make(chan *amqp.Error))
		if !ok {
			r.log.Error("failed notifying rabbitMQ channel. Reconnecting...")
		}
		r.log.Error("rabbitmq connection closed unexpectedly. Reconnecting...")

		for {

			err := r.Connect()

			if err == nil {
				r.log.Info("rabbitMQ reconnect success")
				break
			}

			r.log.Error("rabbitmq reconnect failed. Retry after 10 seconds", sl.Err(err))
			time.Sleep(10 * time.Second)
		}

	}
}

func (r *Rabbit) InitQueue(ch *amqp.Channel, queue string) error {

	dlQueue := queue + "_dead"

	_, err := ch.QueueDeclare(
		dlQueue,
		true,
		false,
		false,
		false,
		amqp.Table{},
	)

	if err != nil {
		return fmt.Errorf("failed to declare dead letter queue: [%w]", err)
	}

	_, err = ch.QueueDeclare(
		queue,
		true,
		false,
		false,
		false,
		amqp.Table{
			"x-dead-letter-exchange":    "",
			"x-dead-letter-routing-key": dlQueue,
			"x-message-ttl":             60480000,
		},
	)

	if err != nil {
		return fmt.Errorf("failed to declare queue: [%w]", err)
	}
	return nil
}

func (r *Rabbit) Consume(ch *amqp.Channel, queue string) (msgs <-chan amqp.Delivery, err error) {
	msgs, err = ch.Consume(
		queue,
		"",
		false,
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		return nil, fmt.Errorf("failed to consume queue: [%w]", err)
	}
	return msgs, nil
}

func (r *Rabbit) Publish(queue string, message []byte) error {
	ch, err := r.Channel()
	if err != nil {
		return fmt.Errorf("failed to open channel: [%w]", err)
	}

	defer ch.Close()

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	err = ch.PublishWithContext(
		ctx,
		"",
		queue,
		false,
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        message,
		},
	)

	if err != nil {
		return fmt.Errorf("failed to publish message: [%w]", err)
	}
	return nil
}

func (r *Rabbit) Connection() *amqp.Connection {
	return r.conn
}

Command

Создадим BaseCommand с общим методом Execute и две реализации: DocumentCommand (LibreOffice) и VideoCommand (MPEG).

// internal/lib/command/command.go
func (bs *BaseCommand) Execute() error {

	if err := bs.validate(); err != nil {
		return fmt.Errorf("failed validate transform task: [%w]", err)
	}

	directory := bs.DownloadDir()

	err := os.MkdirAll(directory, 0755)

	if err != nil {
		return fmt.Errorf("error creating directory [%s]: [%w]", directory, err)
	}

	filePath := bs.genTmpFilePath(directory)

	err = retry.Do(
		func() error {
			return bs.uploader.Download(bs.task.File, filePath, bs.MaxSize())
		},
		retry.Attempts(3),
		retry.OnRetry(func(n uint, err error) {
			time.Sleep(1 * time.Second)
		}),
	)

	bs.uploader.AddFileToDelete(filePath)

	defer bs.uploader.DeleteFiles()

	if err != nil {
		return fmt.Errorf("error download file [%s]: [%w]", bs.task.File, err)
	}

	bs.file = filePath

	for _, format := range bs.task.Formats {

		if _, ok := bs.files[format]; ok {
			continue
		}
		pre, err := bs.preConvert(format, filePath)
		if err != nil {
			return err
		}
		if pre {
			continue
		}

		convertedFile, err := bs.transform(format, filePath)
		bs.uploader.AddFileToDelete(convertedFile)
		if err != nil {
			return fmt.Errorf("error transform file [%s] to [%s]: [%w]", bs.task.File, format, err)
		}
		bs.files[format] = convertedFile
	}

	bs.uploader.SetFiles(bs.files)

	err = bs.uploader.UploadFiles()
	if err != nil {
		return fmt.Errorf("error uploading files: [%w]", err)
	}

	err = bs.uploader.Complete()
	if err != nil {
		return fmt.Errorf("failed complete: [%w]", err)
	}
	return nil
}

Consumer

В текущей реализации запускаем 3 горутины на каждую очередь, в одном контейнере.

Настроим graceful завершение (с таймаутом в 5 минут) и реконнект к rabbitmq (в случае разрыва соединения обработчики будут подключаться заново и после восстановления соединения продолжат работу в штатном режиме)

// cmd/consumer/main.go
func main() {

	cfg := config.MustLoad()

	logger := sl.SetupLogger(cfg.Env)

	rabbit := rabbitmq.New(logger, cfg.Rabbit)

	conErr := rabbit.Connect()
	if conErr != nil {
		log.Fatalf("failed connect to RabbitMQ with start %v", conErr)
	}

	go rabbit.Reconnect()

	cancelCtx, cancel := context.WithCancel(context.Background())
	wg := &sync.WaitGroup{}
	for i := 0; i <= 3; i++ {
		for _, queue := range queues {

			go func() {
				uniqId := fmt.Sprintf("%s_%d", queue, i)
				logger.Info("start consumer", slog.String("queue", queue))
				wg.Add(1)
			done:
				for {
					time.Sleep(10 * time.Second)
					ch, err := rabbit.Channel()
					if err != nil {
						logger.Error("failed to open channel. Retry", slog.String("queue", queue), sl.Err(err))
						continue
					}
					defer ch.Close()

					err = rabbit.InitQueue(ch, queue)

					if err != nil {
						logger.Error("failed init queue. Retry", slog.String("queue", queue), sl.Err(err))
						continue
					}

					logger.Info("success init queue", slog.String("queue", queue))
					msgs, err := rabbit.Consume(ch, queue)

					if err != nil {
						logger.Error("failed consume. Retry", slog.String("queue", queue), sl.Err(err))
						continue
					}

					logger.Info("success consume. Waiting messages", slog.String("queue", queue))
				closed:
					for {
						select {
						case <-cancelCtx.Done():
							break done
						default:
						}
						select {
						case d, ok := <-msgs:
							if !ok {
								logger.Info("channel closed", slog.String("queue", queue))
								break closed
							}
							handleMessage(d, logger, cfg, uniqId)
						default:
						}
                        time.Sleep(1 * time.Second)
					}
				}
				wg.Done()
			}()

		}
	}
	waitCh := make(chan struct{})
	ch := make(chan os.Signal)
	signal.Notify(ch, syscall.SIGTERM, syscall.SIGQUIT, syscall.SIGINT, syscall.SIGKILL)

	<-ch
	logger.Info("receive a shutdown signal")
	go func() {
		logger.Info("cancel, wait consumer")
		cancel()
		wg.Wait()
		close(waitCh)
	}()

	select {
	case <-waitCh:
		logger.Info("graceful shutdown")
	case <-time.After(5 * time.Minute):
		logger.Info("shutdown before 5 minutes timeout")
	}

}

Логирование

Для логирования используем библиотеку slog, для группировки логов используем параметр request_id.

Dead Letter Queue

Для каждой очереди инициализируем одноимённую очередь с префиксом dead.

В случае ошибки конвертации делаем reject, и задание попадает в dead очередь, для последующего анализа. По умолчанию время жизни сообщения - неделя.

Где посмотреть?

Сервер разместил на GitHub, текущая версия 1.0.0. В дальнейшем возможно будет обновляться. Инструкция по развертыванию прилагается.

Итого

Надеюсь, это решение упростит отказ от CentOS там, где это необходимо. При возникновении проблем создавайте issue на GitHub.