Ресурсы функции

То, что функция в один момент времени обрабатывает один запрос, а после выполнения инстанс функции подчищает все ресурсы, позволяет не задумываться о корректном закрытии таких ресурсов. Например, можно спокойно не закрыть открытый на чтение временный файл. И в этом действительно не будет никаких проблем, до тех пор пока в сервис не придет нагрузка.

В этом случае инстансы начнут переиспользоваться: после завершения обработки одного запроса в тот же инстанс будет передаваться на выполнение следующий запрос. С одной стороны это хорошо — сервису Cloud Functions не нужно пересоздавать инстансы, а значит пользователю не нужно будет ждать лишнее время «холодного старта». С другой стороны, именно в этот момент могут всплыть проблемы связанные с некорректной работой с ресурсами.

Кеширование данных в /tmp

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

Например, в функцию приходит запрос от user1 вы получаете его из базы и складываете эту информацию в файл /tmp/user, в надежде, что в следующий раз сможете сэкономить в этом месте время. Но вот в следующий раз приходит запрос от user2, и если вы не проверили, что в кеше у вас данные другого пользователя, то вы вернете пользователю user2 данные user1.

Поэтому если вы и кешируете данные в /tmp, то не забывайте добавлять к ключу какой-то идентификатор пользователя.

Также не стоит забывать, что в /tmp есть ограничение на размер файлов по умолчанию в 512 МБ.

Открытие файлов

Лимит на количество открытых файлов в инстансе функции — 4096. Если вы открываете файлы в цикле, то вам нужно быть уверенным, что вы закрываете их после использования.

Убедиться в размере лимита на количество открытых файлов можно проверив значение ulimit -n внутри функции. А чтобы убедиться, что лимит не может быть превышен, можно воспользоваться следующим кодом:

import os
import subprocess


def handler(event, context):
    stdout = check_ulimit()
    print(stdout)
    for i in range(100000):
        os.open(f"/tmp/f{i}", os.O_CREAT | os.O_WRONLY)

def check_ulimit():
    command = ['sh', '-c', 'ulimit -n']
    process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    stdout, _ = process.communicate()
    return stdout.decode()

В ответе на запрос вы увидите значение 4096, что означает, что лимит на количество открытых файлов в инстансе функции равен 4096. А также функция вернет ошибку:

{
    "errorMessage": "[Errno 24] Too many open files: '/tmp/f4091'",
    "errorType": "OSError",
    "stackTrace": [
        "  File \"/function/runtime/runtime.py\", line 231, in handle_event\n",
        "  File \"/function/code/index.py\", line 10, in handler\n"
    ]
}

Такое же поведение будет и в Go:

{
    "errorMessage": "open /tmp/f4088: too many open files",
    "errorType": "UserCodeError"
}

Видно, что у нас получилось что в примере Python открыть 4091 файл, а вот 4092 файл уже вызвал ошибку. Разница в количестве файлов между Python и Go несущественна и объясняется различиями в реализации рантаймов.

Корректным решением работы с файлами в Python будет использование менеджера контекста with:

import os


def handler(event, context):
    for i in range(100000):
        with open(f"/tmp/f{i}", "w") as f:
            f.write("Hello, World!")

Таким образом, файл будет закрыт автоматически после завершения блока with.

В Go для закрытия файла используется метод Close:

package main

import (
    "fmt"
    "os"
)

func Handler(event map[string]interface{}) error {
    for i := 0; i < 100000; i++ {
        f, err := os.Create(fmt.Sprintf("/tmp/f%d", i))
        if err != nil {
            return err
        }
        f.WriteString("Hello, World!")
        f.Close()
    }
    return nil
}

Стоит отметить, что из-за особенностей работы defer в цикле, в данном случае использование defer f.Close() не рекомендуется, так как это может привести к превышению лимита на количество открытых файлов. Но если вынести тело цикла в отдельную функцию, то можно использовать defer:

package main

import (
    "fmt"
    "os"
)

func createFile(i int) error {
    f, err := os.Create(fmt.Sprintf("/tmp/f%d", i))
    defer f.Close()
    if err != nil {
        return err
    }
    f.WriteString("Hello, World!")
    return nil
}

func Handler(event map[string]interface{}) error {
    for i := 0; i < 100000; i++ {
        err := createFile(i)
        if err != nil {
            return err
        }
    }
    return nil
}

Inodes

Но 100 тысяч файлов вам создать все равно не удастся. Вот такую ошибку вы получите запустив код на Python, приведенный выше:

{
    "errorMessage": "[Errno 28] No space left on device: '/tmp/f32758'",
    "errorType": "OSError",
    "stackTrace": [
        "  File \"/function/runtime/runtime.py\", line 231, in handle_event\n    result = h(r.event, r.context)\n",
        "  File \"/function/code/index.py\", line 6, in handler\n    with open(f\"/tmp/f{i}\", \"w\") as f:\n"
    ]
}

Аналогичная ошибка будет и в Go.

{
    "errorMessage": "open /tmp/f32758: no space left on device",
    "errorType": "UserCodeError"
}

Inodes в /tmp закончились. По умолчанию их 32768. Проверить можно командой df -i:

Filesystem     Inodes IUsed IFree IUse% Mounted on
/var/run         5100    10  5090    1% /run
/dev/vda        32768 32768     0  100% /tmp
tmpfs            5100     1  5099    1% /function/storage
cgroup-sysfs     5100     2  5098    1% /sys/fs/cgroup

Этот лимит вам тоже стоит учитывать, если вы планируете использовать /tmp для кеширования маленьких файлов. Так как, если вы будете исходить из оценок на основе только размера файлов, то вы можете столкнуться с проблемой нехватки inodes.

Примонтированные S3-бакеты

Для бакетов, примонтированных в /function/storage, нет ограничения на количество inodes. df -i покажет, что 1000000000 свободных inodes:

'mount-test:    1000000000     0 1000000000    0% /function/storage/a'

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

Однако, стоит помнить, что доступ к файлам в бакете будет медленнее, чем к файлам в /tmp. Поэтому, если ваши файлы должны быть доступны быстро, то лучше использовать /tmp.

На то, чтобы создать 100 тысяч файлов из примера выше, мне понадобилось 338 секунд.

Ограничение на количество открытых файлов не зависит, где находятся файлы: в /tmp или в примонтированном бакете. Поэтому для бакета сохраняется тот лимит.

{
    "errorMessage": "[Errno 24] Too many open files: '/function/storage/a/f4091'",
    "errorType": "OSError",
    "stackTrace": [
        "  File \"/function/runtime/runtime.py\", line 231, in handle_event\n",
        "  File \"/function/code/index.py\", line 7, in handler\n"
    ]
}

Сетевые соединения

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

Давайте попробуем отрыть 4096 сокетов. Чтобы пример был чуть более интересным, я буду использовать пытаться в цикле послать сообщения в YMQ, используя AWS SDK.

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"sync"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/service/sqs"
)

type Queue struct {
	Client *sqs.Client
	Name   string
	url    string
}

func New(ctx context.Context, region, url string) (*sqs.Client, error) {
	conf, err := config.LoadDefaultConfig(
		ctx,
		config.WithRegion(region),
		config.WithEndpointResolverWithOptions(aws.EndpointResolverWithOptionsFunc(
			func(service, region string, options ...interface{}) (aws.Endpoint, error) {
				return aws.Endpoint{
					PartitionID:   "yc",
					URL:           url,
					SigningRegion: region,
				}, nil
			},
		)),
	)
	if err != nil {
		return nil, err
	}
	return sqs.NewFromConfig(conf), nil
}

func NewQueue(ctx context.Context, region, endpoint string, name string) (*Queue, error) {
	client, err := New(ctx, region, endpoint)
	if err != nil {
		return nil, err
	}
	return &Queue{Client: client, Name: name}, nil
}

func (q *Queue) getURL(ctx context.Context, name string) (string, error) {
	input := sqs.GetQueueUrlInput{
		QueueName: aws.String(name),
	}
	output, err := q.Client.GetQueueUrl(ctx, &input)
	if err != nil {
		return "", err
	}
	return aws.ToString(output.QueueUrl), nil
}

func (q *Queue) Send(ctx context.Context, message string) error {
	if q.url == "" {
		url, err := q.getURL(ctx, q.Name)
		if err != nil {
			return err
		}
		q.url = url
	}
	input := sqs.SendMessageInput{
		QueueUrl:    aws.String(q.url),
		MessageBody: aws.String(message),
	}
	_, err := q.Client.SendMessage(ctx, &input)
	return err
}

func (q *Queue) SendJSON(ctx context.Context, j any) error {
	data, err := json.Marshal(j)
	if err != nil {
		return err
	}
	return q.Send(ctx, string(data))
}

func Handler(event map[string]interface{}) error {
	ctx := context.Background()
	var wg sync.WaitGroup
	errCh := make(chan error, 4096) // Buffer size is equal to the number of goroutines

	go func() {
		for {
			select {
			case err, ok := <-errCh:
				if ok {
					// Handle error
					fmt.Println(err)
				} else {
					// If the channel is closed and drained
					return
				}
			}
		}
	}()

	for i := 0; i < 4096; i++ {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			q, err := NewQueue(
				ctx,
				"ru-central1",
				"https://message-queue.api.cloud.yandex.net",
				"test",
			)
			if err != nil {
				errCh <- err
				return
			}

			data := map[string]interface{}{
				"messageId": i,
			}
			err = q.SendJSON(ctx, data)
			if err != nil {
				errCh <- err
				return
			}
		}(i)
	}
	wg.Wait()
	close(errCh) // Close the error channel when all goroutines have completed
	return nil
}

Так как мы догадываемся, что у нас возникнет проблема с лимитом на количество открытых файлов, то количество итераций в цикле возьмем равным 4096. После запуска функции мы получим ошибку:

operation error SQS: GetQueueUrl, exceeded maximum number of attempts, 3, https response error StatusCode: 0, RequestID: , request send failed, Post "https://message-queue.api.cloud.yandex.net/": dial tcp: lookup message-queue.api.cloud.yandex.net: too many open files

Чтобы eё исправить, во-первых, нам нужно вынести из цикла создание клиента SQS, определение URL очереди. Потому что на этот вызов есть довольно жесткие лимиты, гораздо жестче, чем на отправку сообщений в очередь. Так как все наши горутины будут использовать один и тот же клиент, то этот клиент и будет управлять всеми соединениями. В этом месте мы перестанем исчерпывать лимит на количество открытых файлов. Однако соединения все равно будут порождать dns-запросы. Чтобы поправить и эту проблему нам нужно будет добавить DNS-кэширование прямо в код нашей функции. Для этого возьмем библиотеку dnscache.

Итого настройка клиента SQS будет выглядеть так:

func New(ctx context.Context, region, url string) (*sqs.Client, error) {
	r := &dnscache.Resolver{}

	t := http.DefaultTransport.(*http.Transport).Clone()
	t.MaxIdleConns = 100
	t.MaxConnsPerHost = 100
	t.MaxIdleConnsPerHost = 100
	t.DialContext = func(ctx context.Context, network string, addr string) (conn net.Conn, err error) {
		host, port, err := net.SplitHostPort(addr)
		if err != nil {
			return nil, err
		}
		ips, err := r.LookupHost(ctx, host)
		if err != nil {
			return nil, err
		}
		for _, ip := range ips {
			var dialer net.Dialer
			conn, err = dialer.DialContext(ctx, network, net.JoinHostPort(ip, port))
			if err == nil {
				break
			}
		}
		return
	}


	conf, err := config.LoadDefaultConfig(
		ctx,
		config.WithRegion(region),
		config.WithEndpointResolverWithOptions(aws.EndpointResolverWithOptionsFunc(
			func(service, region string, options ...interface{}) (aws.Endpoint, error) {
				return aws.Endpoint{
					PartitionID:   "yc",
					URL:           url,
					SigningRegion: region,
				}, nil
			},
		)),
		config.WithHTTPClient(&http.Client{
			Transport: t,
		}),
	)
	if err != nil {
		return nil, err
	}
	return sqs.NewFromConfig(conf), nil
}

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

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

Заключение

В этой статье мы рассмотрели, как можно случайно исчерпать ресурсы функции. Важно помнить, что в инстансе функции действительно есть ограничения на количество открытых файлов и inodes, и аккуратно работать с ресурсами. Также важно помнить, что сетевые соединения тоже являются файлами, и их тоже нужно закрывать после использования.